28 |
|
|
29 |
package org.nongnu.storm.impl; |
package org.nongnu.storm.impl; |
30 |
import org.nongnu.storm.*; |
import org.nongnu.storm.*; |
|
import org.nongnu.storm.headers.*; |
|
31 |
import org.nongnu.storm.util.ByteArrayKey; |
import org.nongnu.storm.util.ByteArrayKey; |
32 |
import java.io.*; |
import java.io.*; |
33 |
import java.util.*; |
import java.util.*; |
46 |
protected Map dbs; |
protected Map dbs; |
47 |
|
|
48 |
protected class TransientBlockOutputStream extends AbstractBlockOutputStream { |
protected class TransientBlockOutputStream extends AbstractBlockOutputStream { |
49 |
protected TransientBlockOutputStream(Header822 header) |
protected TransientBlockOutputStream(String contentType) |
50 |
throws IOException { |
throws IOException { |
51 |
super(new ByteArrayOutputStream(), header); |
super(new ByteArrayOutputStream(), contentType); |
52 |
} |
} |
53 |
|
|
54 |
public Block makeBlock() throws IOException { |
public Block makeBlock() throws IOException { |
55 |
ByteArrayOutputStream baos = (ByteArrayOutputStream)out; |
ByteArrayOutputStream baos = (ByteArrayOutputStream)out; |
56 |
block = new TransientBlock(makeIdFromDigest(), |
block = new TransientBlock(makeIdFromDigest(), |
57 |
baos.toByteArray(), header); |
baos.toByteArray()); |
58 |
blocks.put(block.getId(), block); |
blocks.put(block.getId(), block); |
59 |
|
|
60 |
added(block); |
added(block); |
64 |
|
|
65 |
protected class TransientBlock extends AbstractBlock { |
protected class TransientBlock extends AbstractBlock { |
66 |
protected byte[] bytes; |
protected byte[] bytes; |
|
protected int headerLength; |
|
67 |
|
|
68 |
protected TransientBlock(BlockId id, byte[] bytes, |
protected TransientBlock(BlockId id, byte[] bytes) |
69 |
Header822 header) throws IOException { |
throws IOException { |
70 |
super(id, header); |
super(id); |
71 |
this.bytes = bytes; |
this.bytes = bytes; |
72 |
|
|
73 |
id.check(bytes); |
id.check(bytes); |
74 |
} |
} |
75 |
|
|
76 |
public InputStream getRawInputStream() throws IOException { |
public InputStream getInputStream() throws IOException { |
77 |
return new ByteArrayInputStream(bytes); |
return new ByteArrayInputStream(bytes); |
78 |
} |
} |
79 |
} |
} |
111 |
return (Block)blocks.get(id); |
return (Block)blocks.get(id); |
112 |
} |
} |
113 |
public void add(Block b) throws IOException { |
public void add(Block b) throws IOException { |
114 |
byte[] bytes = org.nongnu.storm.util.CopyUtil.readBytes(b.getRawInputStream()); |
byte[] bytes = org.nongnu.storm.util.CopyUtil.readBytes(b.getInputStream()); |
115 |
|
|
116 |
BlockId id = b.getId(); |
BlockId id = b.getId(); |
117 |
id.check(bytes); |
id.check(bytes); |
118 |
|
|
119 |
InputStream is = new ByteArrayInputStream(bytes); |
Block block = new TransientBlock(id, bytes); |
|
Header822 header = Headers822.readHeader(is); |
|
|
is.close(); |
|
|
|
|
|
Block block = new TransientBlock(id, bytes, header); |
|
120 |
blocks.put(id, block); |
blocks.put(id, block); |
121 |
|
|
122 |
added(block); |
added(block); |
129 |
} |
} |
130 |
public BlockOutputStream getBlockOutputStream(String contentType) |
public BlockOutputStream getBlockOutputStream(String contentType) |
131 |
throws IOException { |
throws IOException { |
132 |
Header822 hdr = new UniqueHeader822(); |
return new TransientBlockOutputStream(contentType); |
|
hdr.add("Content-Type", contentType); |
|
|
return new TransientBlockOutputStream(hdr); |
|
|
} |
|
|
public BlockOutputStream getBlockOutputStream(Header822 hdr) |
|
|
throws IOException { |
|
|
return new TransientBlockOutputStream(new VerbatimHeader822(hdr)); |
|
133 |
} |
} |
134 |
|
|
135 |
protected DB getDB(IndexType indexType) { |
protected DB getDB(IndexType indexType) { |