public void closeAll() {

        // Close any active streams
        for (InterceptedStream<T> stream : streams.values())
            close(stream.getStream());

        // Remove now-useless references
        streams.clear();

    }

    /**
     * Blocks until the given stream is closed, or until another stream with
     * the same index replaces it.
     *
     * @param stream
     *     The stream to wait for.
     */
    public void waitFor(InterceptedStream<T> stream) {

        T underlyingStream = stream.getStream();

        // Wait for stream to close
        synchronized (underlyingStream) {
            while (streams.get(stream.getIndex()) == stream) {
                try {
                    underlyingStream.wait(STREAM_WAIT_TIMEOUT);
                }
                catch (InterruptedException e) {
                    // Ignore
                }
            }
        }

    }

    /**
     * Returns the stream stored in this map under the given index.
     *
     * @param index
     *     The index of the stream to return.
     *
     * @return
     *     The stream having the given index, or null if no such stream is
     *     stored within this map.
     */
    public InterceptedStream<T> get(String index) {
        return streams.get(index);
    }

    /**
     * Adds the given stream to this map, storing it under its associated
     * index. If another stream already exists within this map having the same
     * index, that stream will be closed and replaced.
     *
     * @param stream
     *     The stream to store within this map.
     */
    public void put(InterceptedStream<T> stream) {

        // Add given stream to map
        InterceptedStream<T> oldStream =
                streams.put(stream.getIndex(), stream);

        // If a previous stream DID exist, close it
        if (oldStream != null)
            close(oldStream.getStream());

    }

}
public void closeAll ( ) { for ( InterceptedStream < T > stream : streams . values ( ) ) close ( stream . getStream ( ) ) ; streams . clear ( ) ; } public void waitFor ( InterceptedStream < T > stream ) { T underlyingStream = stream . getStream ( ) ; synchronized ( underlyingStream ) { while ( streams . get ( stream . getIndex ( ) ) == stream ) { try { underlyingStream . wait ( STREAM_WAIT_TIMEOUT ) ; } catch ( InterruptedException e ) { } } } } public InterceptedStream < T > get ( String index ) { return streams . get ( index ) ; } public void put ( InterceptedStream < T > stream ) { InterceptedStream < T > oldStream = streams . put ( stream . getIndex ( ) , stream ) ; if ( oldStream != null ) close ( oldStream . getStream ( ) ) ; } }
Removes and closes all streams stored within this map, logging any errors
that occur during closure, and unblocking any in-progress calls to
waitFor().