//============================================================================== // OutputStreamMerger.java //============================================================================== package tribble.io; import java.io.IOException; import java.io.OutputStream; import java.lang.ArrayIndexOutOfBoundsException; import java.lang.String; import java.lang.System; import java.lang.UnsupportedOperationException; import java.util.ArrayList; /******************************************************************************* * Output stream funnel. * Merges the output of one or more output streams into the same underlying * output stream. * *

* Data written to any of the source streams for this object is merged together * and written to the single output (sink) stream of this object. * This acts like a funnel, combining the output data from several source streams * into a single output stream. This can be useful when several threads are * writing to a single text file simultaneously. * * *

Source code:: * http://david.tribble.com/src/java/tribble/io/OutputStreamMerger.java *
Documentation:: * http://david.tribble.com/docs/tribble/io/OutputStreamMerger.html *

* * * @version $Revision: 1.2 $ $Date: 2008/09/26 14:29:36 $ * @since 2008-09-25 * @author David R. Tribble (david@tribble.com) *

* Method {@link #getSource getSource()} is called to create one or more * source streams for this object's output stream. All data written to the * source streams is merged and written to the single underlying output * stream. * * @param out * The underlying output stream, to which all the data written to the source * streams associated with this object is written. * * @since 1.1, 2008-09-25 */ public OutputStreamMerger(OutputStream out) { m_out = out; } // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - // Methods /*************************************************************************** * Create a source output stream that writes merged data to the underlying * output stream of this object. * *

* Note that each call to this method creates a new independent data source * for the underlying output stream. * * @param eoln * End-of-line character code for the buffered output stream. * All output written to the stream is buffered by lines, and is only written * when the internal line buffer becomes full or when an end-of-line * character code is written. * If this is equal to {@link #EOLN_NATIVE}, the native end-of-line character * for the underlying operating system is used. * Is this is equal to {@link #EOLN_NONE}, no end-of-line character is used, * and output is written only when the internal line buffer becomes full. *

* * @param autoFlush * If true, the underlying output stream is flushed (see {@link #flush()}) * whenever the flush() method of the source stream is called; * otherwise, only the internal line buffer of the source stream is flushed * to the underlying output stream. *

* * @param bufSize * Size (in bytes) of the internal line buffer. * If this is zero or negative, the default size of 1024 bytes is used. * * @return * A source output stream. Data written to the output stream is merged with * data written to the same underlying output stream shared by other source * streams. * * @since 1.2, 2008-09-26 */ public OutputStream getSource(int eoln, boolean autoFlush, int bufSize) { // Create and add a source stream to this output object return new SourceStream(this, eoln, autoFlush, bufSize); } /*************************************************************************** * Create a source output stream that writes merged data to the underlying * output stream of this object. * This is equivalent to the call: *

    *    {@link #getSource(int, boolean, int) getSource}({@link #EOLN_NATIVE}, false, 0)

* * @return * A source output stream. Data written to the output stream is merged with * data written to the same underlying output stream shared by other source * streams. * * @since 1.2, 2008-09-26 */ public OutputStream getSource() { // Create and add a source stream to this output object return getSource(EOLN_NATIVE, false, 0); } /*************************************************************************** * Flush any pending output to the underlying output stream. * * @throws IOException * Thrown if an I/O (write) error occurs on the underlying output stream. * * @since 1.1, 2008-09-25 */ public void flush() throws IOException //overrides java.io.OutputStream { // Sanity check if (m_out == null) throw new IOException("Output stream is closed"); // Flush any pending output m_out.flush(); } /*************************************************************************** * Close the underlying output stream. * Subsequent attempts to write data to any of the source streams associated * with this object will cause exceptions. * *

* Note that this method can be called multiple times without any ill * effects. * * @throws IOException * Thrown if an I/O error occurs on the underlying output stream. * * @since 1.1, 2008-09-25 */ public void close() throws IOException //overrides java.io.OutputStream { // Sanity check if (m_out == null) return; // Close the underlying stream m_out.close(); // Disconnect the underlying stream m_out = null; } /*************************************************************************** * Write a set of bytes to the underlying output stream. * This method is called only by the methods of an helper inner class. * * * @param buf * An array of bytes to be written. *

* * @param off * Index of the first byte in buf to write. *

* * @param len * Number of bytes in buf to write. * * @throws IOException * Thrown if an I/O (write) error occurs on the underlying output stream. * * @since 1.1, 2008-09-25 */ void writeBytes(/*const*/ byte[] buf, int off, int len) throws IOException { // Sanity check if (m_out == null) throw new IOException("Output stream is closed"); // Write the data bytes to the underlying output stream m_out.write(buf, off, len); } // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - // Inner classes /*************************************************************************** * Source output stream for a {@link OutputStreamMerger}. * Data written to this source stream is merged with data written to all * other source streams attached to the same {@link OutputStreamMerger} * object. * *

* Note that the methods of this class are not synchronized. * * *

Source code:: * http://david.tribble.com/src/java/tribble/io/OutputStreamMerger.java *
Documentation:: * http://david.tribble.com/docs/tribble/io/OutputStreamMerger.SourceStream.html *

* * * @version $Revision: 1.2 $ $Date: 2008/09/26 14:29:36 $ * @since {@link OutputStreamMerger} 1.1, 2008-09-25 * @author David R. Tribble (david@tribble.com) *

* Copyright Š2008 by David R. Tribble, all rights reserved.
* Permission is granted to any person or entity except those designated by * by the United States Department of State as a terrorist, or terrorist * government or agency, to use and distribute this source code provided * that the original copyright notice remains present and unaltered. */ private static class SourceStream extends java.io.OutputStream { static final String REV = "@(#)tribble/io/OutputStreamMerger.java $Revision: 1.2 $ $Date: 2008/09/26 14:29:36 $\n"; // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - // Constants /** Default output line buffer size. */ private static final int DFL_BUFSIZE = 1024; // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - // Variables /** Underlying merged output (sink) stream. */ private OutputStreamMerger m_out; /** Output line buffer. */ private byte[] m_buf; /** Output line buffer index. */ private int m_next; /** End-of-line flush character (see OutputStreamMerger.EOLN_XXX). */ private int m_eoln; /** {@link #flush()} also flushes the underlying output stream. */ private boolean m_autoFlush; // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - // Constructors /*********************************************************************** * Constructor. * * @param out * The underlying output (sink) stream for all data written to this * source output stream. *

* * @param eoln * End-of-line character code for the buffered output stream. * All output written to the stream is buffered by lines, and is only * written when the internal line buffer becomes full or when an * end-of-line character code is written. * If this is equal to {@link OutputStreamMerger#EOLN_NATIVE}, the native * end-of-line character for the underlying operating system is used. * Is this is equal to {@link OutputStreamMerger#EOLN_NONE}, no * end-of-line character is used, and output is written only when the * internal line buffer becomes full. *

* * @param autoFlush * If true, the underlying output stream is flushed whenever the * {@link #flush flush()} method of this source stream is called; * otherwise, only the internal line buffer of the source stream is * flushed to the underlying output stream. *

* * @param bufSize * Size (in bytes) of the internal line buffer. * If this is zero or negative, the default size of 1024 bytes is used. * * @since 1.2, 2008-09-26 */ SourceStream(OutputStreamMerger out, int eoln, boolean autoFlush, int bufSize) { m_out = out; m_eoln = eoln; m_autoFlush = autoFlush; m_buf = new byte[bufSize > 0 ? bufSize : DFL_BUFSIZE]; } // - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - // Methods /*********************************************************************** * Flush any pending output to the output stream. * Forces any pending output written to this stream to be written to the * underlying output {@link OutputStreamMerger} it is attached to. * This occurs regardless of whether or not the output is line-buffered. * *

* If auto-flush is set, the underlying {@link OutputStreamMerger} is * also flushed. * * @throws IOException * Thrown if an I/O (write) error occurs on the underlying output stream. * * @since 1.1, 2008-09-25 */ public void flush() throws IOException //overrides java.io.OutputStream { // Sanity check if (m_out == null) throw new IOException("Output stream is closed"); if (m_next <= 0) return; // Flush all pending output to the underlying stream synchronized (m_out) { m_out.writeBytes(m_buf, 0, m_next); m_next = 0; if (m_autoFlush) m_out.flush(); } } /*********************************************************************** * Close this source output stream. * Any pending output is first flushed to the underlying output stream. * Subsequent attempts to write data to this output stream will cause * exceptions to be thrown. * *

* Note that this method can be called multiple times without any ill * effects. * *

* Note that the underlying output stream is not closed. * * @throws IOException * Thrown if an I/O error occurs on the underlying output stream. * * @since 1.1, 2008-09-25 */ public void close() throws IOException //overrides java.io.OutputStream { // Sanity check if (m_out == null) return; // Flush any pending output to the underlying stream flush(); // Disconnect the underlying stream m_out = null; m_buf = null; } /*********************************************************************** * Write a byte to the underlying output stream. * * @param b * An output byte. * * @throws IOException * Thrown if an I/O (write) error occurs on the underlying output stream. * * @since 1.1, 2008-09-25 */ public void write(int b) throws IOException //overrides java.io.OutputStream { // Sanity check if (m_out == null) throw new IOException("Output stream is closed"); // Check for a full output line buffer if (m_next >= m_buf.length) flush(); // Add a byte to the output line buffer m_buf[m_next++] = (byte) b; } /*********************************************************************** * Write a set of bytes to the underlying output stream. * This is equivalent to the call: *

        *    {@link #write(byte[], int, int) write}(buf, 0, buf.length)

* * @param buf * An array of bytes to be written. * * @throws IOException * Thrown if an I/O (write) error occurs on the underlying output stream. * * @since 1.1, 2008-09-25 */ public void write(/*const*/ byte[] buf) throws IOException //overrides java.io.OutputStream { write(buf, 0, buf.length); } /*********************************************************************** * Write a set of bytes to the underlying output stream. * * @param buf * An array of bytes to be written. *

* * @param off * Index of the first byte in buf to write. *

* * @param len * Number of bytes in buf to write. * * @throws IOException * Thrown if an I/O (write) error occurs on the underlying output stream. * * @since 1.1, 2008-09-25 */ public void write(/*const*/ byte[] buf, int off, int len) throws IOException //overrides java.io.OutputStream { // Sanity check if (m_out == null) throw new IOException("Output stream is closed"); // Append the data bytes to the output line buffer while (len > 0) { // Check for a full line buffer if (m_next >= m_buf.length) flush(); // Write some of the data to the end of the line buffer while (len > 0 && m_next < m_buf.length) { int b; // Append the output byte to the end of the line buffer b = buf[off++] & 0xFF; m_buf[m_next++] = (byte) b; len--; // Flush at each end of line if (b == m_eoln) flush(); } } } } } // End OutputStreamMerger.java