/* AsyncSetCollector.java * * Copyright (c) 2003, Benja Fallenstein * * You may use and distribute under the terms of either the GNU Lesser * General Public License, either version 2 of the license or, * at your choice, any later version. Alternatively, you may use and * distribute under the terms of the XPL. * * See the LICENSE.lgpl and LICENSE.xpl files for the specific terms of * the licenses. * * This software 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 README * file for more details. * */ /* * Written by Benja Fallenstein */ package gzz.storm.impl; import gzz.storm.*; import java.util.*; /** XXX */ public class AsyncSetCollector implements SetCollector { /** The set of elements that have been read so far. * Must always be synchronized... */ protected final Set set = new HashSet(); /** The time this object was created. * Used by getAge(). */ protected final long creationTime; /** Zero for working, >0 for complete, <0 for timeout. */ protected int state; protected Set listeners = new HashSet(); public AsyncSetCollector() { this.creationTime = System.currentTimeMillis(); this.state = 0; } protected void receive(Object o) { if(state != 0) throw new IllegalStateException("Cannot receive more elements"); set.add(o); for(Iterator i=listeners.iterator(); i.hasNext();) { CollectionListener l = (CollectionListener)i.next(); if(!l.item(o)) i.remove(); } } protected synchronized void finish(boolean timeout) { state = timeout ? -1 : 1; notifyAll(); } // Implementation of SetCollector public long getAge() { return System.currentTimeMillis() - creationTime; } public boolean isReady() { return state != 0; } public boolean isComplete() { if(state > 0) return true; else if(state < 0) return false; else throw new IllegalStateException("Collector not ready"); } public Collector block() { return blockSet(); } public synchronized SetCollector blockSet() { if(state != 0) return this; try { wait(); } catch(InterruptedException e) { } return this; } public void addCollectionListener(CollectionListener l) { listeners.add(l); synchronized(set) { for(Iterator i=set.iterator(); i.hasNext();) { if(l.item(i.next())) return; } } } public int size() { return set.size(); } public boolean isEmpty() { return set.isEmpty(); } public boolean contains(Object o) { return set.contains(o); } public Iterator iterator() { return set.iterator(); } public Object[] toArray() { return set.toArray(); } public Object[] toArray(Object[] a) { return set.toArray(a); } public boolean add(Object o) { throw new UnsupportedOperationException("Unmodifiable collector"); } public boolean remove(Object o) { throw new UnsupportedOperationException("Unmodifiable collector"); } public boolean containsAll(Collection c) { return set.containsAll(c); } public boolean addAll(Collection c) { throw new UnsupportedOperationException("Unmodifiable collector"); } public boolean removeAll(Collection c) { throw new UnsupportedOperationException("Unmodifiable collector"); } public boolean retainAll(Collection c) { throw new UnsupportedOperationException("Unmodifiable collector"); } public void clear() { throw new UnsupportedOperationException("Unmodifiable collector"); } public boolean equals(Object o) { return set.equals(o); } public int hashCode() { return set.hashCode(); } }