/*
TransientPool.java
*
* Copyright (c) 2002, Benja Fallenstein and Anton Feldmann
*
* This file is part of Storm.
*
* Storm is free software; you can redistribute it and/or modify it under
* the terms of the GNU Lesser General Public License as published by
* the Free Software Foundation; either version 2 of the License, or
* (at your option) any later version.
*
* Storm is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
* or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General
* Public License for more details.
*
* You should have received a copy of the GNU Lesser General
* Public License along with Storm; if not, write to the Free
* Software Foundation, Inc., 59 Temple Place, Suite 330, Boston,
* MA 02111-1307 USA
*
*
*/
/*
* Written by Benja Fallenstein and Anton Feldmann
*/
package org.nongnu.storm.impl;
import org.nongnu.storm.*;
import org.nongnu.storm.headers.*;
import org.nongnu.storm.util.ByteArrayKey;
import java.io.*;
import java.util.*;
/** A StormPool whose contents are exclusively stored in memory.
*/
public class TransientPool extends AbstractLocalPool {
/** The blocks in this pool.
* A mapping from BlockId
objects
* to TransientBlock
s.
*/
protected Map blocks = new HashMap();
/** The DB objects by index type URI.
*/
protected Map dbs;
protected class TransientBlockOutputStream extends AbstractBlockOutputStream {
protected TransientBlockOutputStream(Header822 header)
throws IOException {
super(new ByteArrayOutputStream(), header);
}
public Block makeBlock() throws IOException {
ByteArrayOutputStream baos = (ByteArrayOutputStream)out;
block = new TransientBlock(makeIdFromDigest(),
baos.toByteArray(), header);
blocks.put(block.getId(), block);
added(block);
return block;
}
}
protected class TransientBlock extends AbstractBlock {
protected byte[] bytes;
protected int headerLength;
protected TransientBlock(BlockId id, byte[] bytes,
Header822 header) throws IOException {
super(id, header);
this.bytes = bytes;
id.check(bytes);
}
public InputStream getRawInputStream() throws IOException {
return new ByteArrayInputStream(bytes);
}
}
protected class TransientDB implements DB {
protected Map mappings = new HashMap();
protected Set getSet(byte[] key) {
ByteArrayKey k = new ByteArrayKey(key);
Set s = (Set)mappings.get(k);
if(s == null) {
s = new HashSet();
mappings.put(k, s);
}
return s;
}
public Collector get(byte[] key) {
return new SimpleSetCollector(new HashSet(getSet(key)));
}
protected void add(Mapping m) {
getSet(m.key).add(m);
}
}
public TransientPool(Set indexTypes) throws IOException {
super(indexTypes);
}
public Block get(BlockId id) throws FileNotFoundException {
if(!blocks.keySet().contains(id))
throw new FileNotFoundException("No such block: "+id);
return (Block)blocks.get(id);
}
public void add(Block b) throws IOException {
byte[] bytes = org.nongnu.storm.util.CopyUtil.readBytes(b.getRawInputStream());
BlockId id = b.getId();
id.check(bytes);
InputStream is = new ByteArrayInputStream(bytes);
Header822 header = Headers822.readHeader(is);
is.close();
Block block = new TransientBlock(id, bytes, header);
blocks.put(id, block);
added(block);
}
public void delete(Block b) {
blocks.keySet().remove(b.getId());
}
public SetCollector getIds() {
return new SimpleSetCollector(new HashSet(blocks.keySet()));
}
public BlockOutputStream getBlockOutputStream(String contentType)
throws IOException {
Header822 hdr = new UniqueHeader822();
hdr.add("Content-Type", contentType);
return new TransientBlockOutputStream(hdr);
}
public BlockOutputStream getBlockOutputStream(Header822 hdr)
throws IOException {
return new TransientBlockOutputStream(new VerbatimHeader822(hdr));
}
protected DB getDB(IndexType indexType) {
if(dbs == null) dbs = new HashMap();
String typeURI = indexType.getIndexTypeURI();
DB db = (DB)dbs.get(typeURI);
if(db == null) {
db = new TransientDB();
dbs.put(typeURI, db);
}
return db;
}
protected void added(Block block) throws IOException {
for(Iterator i = indexTypes.iterator(); i.hasNext();) {
IndexType it = (IndexType)i.next();
Set mappings = it.getMappings(block);
for(Iterator j = mappings.iterator(); j.hasNext();) {
Mapping m = (Mapping)j.next();
if(!m.block.equals(block.getId()))
throw new Error("Wrong block in mapping: "+m);
((TransientDB)getDB(it)).add(m);
}
}
}
}