# StreamConnector.java

    /*
    ** Module   : StreamConnector.java
    ** Abstract : connects two streams and forwards contents of one to the other
    **
    ** Copyright (c) 2005-2022, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- -T- --JPRM-- ----------------Description-----------------
    ** 001 GES 20050629 NEW   @21567 Created initial version which provides a
    **                               mechanism to connect two streams and
    **                               forward the contents from one to the other.
    ** 002 GES 20060216 CHG   @24668 Safer close semantics.
    ** 003 TJD 20220504              Java 11 compatibility minor changes
    */
    /*
    ** This program is free software: you can redistribute it and/or modify
    ** it under the terms of the GNU Affero General Public License as
    ** published by the Free Software Foundation, either version 3 of the
    ** License, or (at your option) any later version.
    **
    ** This program 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 Affero General Public License for more details.
    **
    ** You may find a copy of the GNU Affero GPL version 3 at the following
    ** location: https://www.gnu.org/licenses/agpl-3.0.en.html
    **
    ** Additional terms under GNU Affero GPL version 3 section 7:
    **
    **   Under Section 7 of the GNU Affero GPL version 3, the following additional
    **   terms apply to the works covered under the License.  These additional terms
    **   are non-permissive additional terms allowed under Section 7 of the GNU
    **   Affero GPL version 3 and may not be removed by you.
    **
    **   0. Attribution Requirement.
    **
    **     You must preserve all legal notices or author attributions in the covered
    **     work or Appropriate Legal Notices displayed by works containing the covered
    **     work.  You may not remove from the covered work any author or developer
    **     credit already included within the covered work.
    **
    **   1. No License To Use Trademarks.
    **
    **     This license does not grant any license or rights to use the trademarks
    **     Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
    **     of Golden Code Development Corporation. You are not authorized to use the
    **     name Golden Code, FWD, or the names of any author or contributor, for
    **     publicity purposes without written authorization.
    **
    **   2. No Misrepresentation of Affiliation.
    **
    **     You may not represent yourself as Golden Code Development Corporation or FWD.
    **
    **     You may not represent yourself for publicity purposes as associated with
    **     Golden Code Development Corporation, FWD, or any author or contributor to
    **     the covered work, without written authorization.
    **
    **   3. No Misrepresentation of Source or Origin.
    **
    **     You may not represent the covered work as solely your work.  All modified
    **     versions of the covered work must be marked in a reasonable way to make it
    **     clear that the modified work is not originating from Golden Code Development
    **     Corporation or FWD.  All modified versions must contain the notices of
    **     attribution required in this license.
    */

    package com.goldencode.p2j.io;

    import java.io.*;

    /**
     * General purpose service that connects two streams and forwards contents
     * of the input stream to the output stream.  The streams are specified by
     * an <code>InputStream</code> and <code>OutputStream</code>.
     * <p>
     * Uses a worker thread to provide this service.  The thread is created and
     * started in the constructor.  This thread copies data while data exists in
     * the input stream and pauses in between tests of whether any data exists
     * to be copied. This polling loop will continue until the close flag is
     * raised (see {@link #close}) or there is an <code>IOException</code> or
     * <code>InterruptedException</code> on the copy thread.  The core
     * implementation of the copying can be found in the {@link #run} method.
     *
     * @author    GES
     */
    public class StreamConnector
    implements Runnable
    {
       /** Size of the byte array used for block copying. */
       private static final int BUF_SIZE = 8192;

       /** Milliseconds to sleep between polls of the input stream. */
       private static final long SLEEP_INTERVAL = 256;

       /** The input stream from which we read. */
       private InputStream in  = null;

       /** The output stream to which we write. */
       private OutputStream out = null;

       /** An object on which the close operations are synchronized. */
       private Object lock = new Object();

       /** Specifies if the copy thread should exit when no data is left. */
       private boolean flag = false;

       /** If <code>true</code> the input stream should be closed on exit. */
       private boolean closeIn = false;

       /** If <code>true</code> the output stream should be closed on exit. */
       private boolean closeOut = false;

       /**
        * Instantiates the connector and starts the copy thread that drives the
        * transfer of data from the input stream to the output stream.
        *
        * @param    in
        *           The input stream from which to read.
        * @param    out
        *           The output stream on which to write.
        * @param    closeIn
        *           If <code>true</code>, at exit the copy thread will close
        *           the input stream, otherwise the stream will not be closed.
        * @param    closeOut
        *           If <code>true</code>, at exit the copy thread will close
        *           the output stream, otherwise the stream will not be closed.
        */
       public StreamConnector(InputStream  in,
                              OutputStream out,
                              boolean      closeIn,
                              boolean      closeOut)
       {
          this.in       = in;
          this.out      = out;
          this.closeIn  = closeIn;
          this.closeOut = closeOut;

          // create and start the thread that drives the copying
          Thread t1 = new Thread(this);
          t1.start();
       }

       /**
        * Sets the close flag to trigger an exit by the copy thread when no data
        * remains in the input stream.
        */
       public void close()
       {
          synchronized (lock)
          {
             if (!flag)
             {
                flag = true;

                try
                {
                   lock.wait();
                }

                catch (InterruptedException ie)
                {
                   // best efforts to wait only
                }
             }
          }
       }

       /**
        * Checks the state of the close flag.  If <code>true</code>, the copy
        * thread will close when no data remains in the input stream.
        */
       public boolean isClosed()
       {
          boolean result = false;

          synchronized (lock)
          {
             result = flag;
             lock.notifyAll();
          }

          return result;
       }

       /**
        * Copies data from the input stream to the output stream as long as
        * the close flag is <code>false</code> or there is data to copy.  On
        * exit, the input and output streams are closed if and only if such a
        * close was specified in the constructor.  This method is called on a
        * new thread which is instantiated in the constructor.  This thread
        * exits when this method is complete.
        * <p>
        * The copy algorithm is based on a block size specified by the member
        * <code>BUF_SIZE</code>.
        * <p>
        * Please note that as of J2SE 1.4.x there are no facilities that provide
        * non-blocking I/O for pipes used between processes.  Both the original
        * <code>java.io</code> and <code>java.nio</code> packages support pipes
        * for use between 2 threads in the same JVM but DO NOT support the
        * ability to set the ends of the pipe using pipes that are externally
        * created.  For this reason, this method uses a polling approach to
        * the detection of available bytes.  This allows the thread to drop out
        * cleanly when the close flag is turned on, even in the case where no
        * data has been written to the pipe in a long time.  The polling interval
        * is specified by the member <code>SLEEP_INTERVAL</code>.
        */
       public void run()
       {
          byte[] buffer = new byte[BUF_SIZE];

          try
          {
             // continue while there is data to copy but if there is no data
             // AND the close flag is true, then we exit; note that the close
             // flag is tested using a synchronized method to ensure thread
             // safety
             while (in.available() > 0 || !isClosed())
             {
                int num = in.available();

                while (num > 0)
                {
                   // read the data as a full block or the total amount avail
                   int obtained = in.read(buffer, 0, Math.min(num, BUF_SIZE));

                   // decrement our total remaining available bytes
                   num -= obtained;

                   // write the data to the output stream
                   out.write(buffer, 0, obtained);
                }
                Thread.sleep(SLEEP_INTERVAL);
             }
          }

          catch (IOException ioe)
          {
             // exit silently
          }

          catch (InterruptedException ie)
          {
             // exit silently
          }

          finally
          {
             try
             {
                synchronized (lock)
                {
                   // ensure that a close() called after this point is safe
                   if (!flag)
                   {
                      flag = true;
                   }
                   else
                   {
                      lock.notifyAll();
                   }
                }

                // close the input stream if requested
                if (closeIn)
                {
                   in.close();
                }

                // close the output stream if requested
                if (closeOut)
                {
                   out.close();
                }
             }

             catch (IOException ioe)
             {
                // best efforts to close, exit silently
             }
          }
       }
    }