# Protocol.java

    /*
    ** Module   : Protocol.java
    ** Abstract : low level driver of the P2J protocol
    **
    ** Copyright (c) 2005-2024, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --JPRM-- ----------------Description-----------------
    ** 001 SIY 20050120   @19501 Created initial version.
    ** 002 SIY 20050207   @19712 Implemented exception delivering scheme.
    **                           Also modified stop() to avoid waiting for
    **                           threads.
    ** 003 SIY 20050211   @19784 Reset stream to exclude backreferencing.
    ** 004 NVS 20050308   @20250 Using NetSocket class instead of SSLSocket.
    **                           The ObjectInput and ObjectOutput streams are
    **                           created in NetSocket.
    ** 005 SIY 20050315   @20385 Organized imports and other minor cleanups.
    **                           Added call to stream flush() method (merge
    **                           from NVS).
    ** 006 SIY 20050329   @20546 Minor changes in member access rights to
    **                           improve performance (change is very small).
    ** 007 GES 20060211   @24533 Minor cleanup and added to-dos to exception
    **                           processing.
    ** 008 NVS 20060616   @27376 Exception processing modified. No more stack
    **                           trace printing.
    ** 009 GES 20060726   @28199 Added support for bidirectional state
    **                           synchronization.
    ** 010 GES 20060731   @28256 Greatly improved performance by reducing the
    **                           overhead of object graph serialization. The
    **                           trick is to serialize the object twice! First
    **                           the object is flattened via serialization to
    **                           a byte[]. Then this byte[] is serialized to
    **                           the socket.  This intermediate step is
    **                           significantly faster than serializing direct
    **                           to the socket! The reading was modified to
    **                           handle the proper restore.
    ** 011 GES 20060802   @28293 Remove attachments before enqueueing because
    **                           REQUEST_SYNC messages tend to reuse the same
    **                           message on the return trip, but just swap
    **                           addresses. Thus the same state sync data was
    **                           sent back when it didn't need to be in the
    **                           case where the attachment would have been
    **                           null (no changes to send).
    ** 012 NVS 20061017   @30464 Implemented ability to restart the session
    **                           without exiting the process.
    ** 013 GES 20061207   @31612 Added logging and isRunning().
    ** 014 NVS 20070411   @32929 Protocol threads are created as part of a
    **                           thread group named "protocol".
    ** 015 ECF 20071118   @35884 Reworked exception handling in reader and
    **                           writer threads.
    ** 016 GES 20081024   @40328 Eliminated use of the NetSocket/Socket and
    **                           instead now the protocol is only dependent
    **                           upon Queue APIs. This reduces dependencies
    **                           between components and fosters better
    **                           encapsulation of the protocol.
    ** 017 ECF 20090205   @42106 Changed to accommodate updated Queue API.
    **                           Queue.stop() method now accepts a boolean
    **                           parameter to optionally switch security
    **                           context to the session's user's context.
    ** 018 GES 20090722   @43334 Better thread names to enhance debugging.
    ** 019 GES 20090723   @43346 Better handling of OOME on reader/writer threads.
    **                           Improved control over logging detail. Synchronized
    **                           reader/writer thread termination processing to
    **                           avoid extra errors occurring do to a race
    **                           condition.
    ** 020 CA  20090902   @43821 Changed logging level for any SocketException's
    **                           caught while the queue is still running to always
    **                           use FINE.
    ** 021 CA  20131121          The sync'ed state processing is moved from Reader/Writer threads to
    **                           the thread which actually sends/receives the message (i.e.
    **                           Conversation thread).
    ** 022 SBI 20151119          Added stopQueue with SilentUnwindException checking.
    ** 023 CA  20190315          No state synchronization is performed for async replies or requests.
    ** 024 IAS 20200722          Refactored to the new NetSocket API.
    ** 025 CA  20200923          Improved performance for objToByteArray and byteArrayToObj - use Externalizable.
    **                           Changed StateSynchronizer to return Externalizable instances.
    ** 026 IAS 20200930          Added JMX counters.
    ** 027 IAS 20201001          Ged rid of byte arrays allocations in objToByteArray().
    **     IAS 20210325          Added SSLException catch
    **     TJD 20220607          Writer/ByteArrayOutputStream memory leak fix for LOBs
    ** 028 GBB 20230512          Logging methods replaced by CentralLogger/ConversionStatus.
    ** 029 GBB 20230620          createThreadName method moved to Utils.
    ** 030 HC  20240222          Enabled JMX on FWD Client.
    ** 031 CA  20240809          Skip using JMX timers when JMX_DEBUG flag is not set.
    */
    /*
    ** 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.net;

    import java.io.*;
    import java.net.*;
    import java.util.logging.*;

    import javax.net.ssl.*;

    import com.goldencode.p2j.jmx.*;
    import com.goldencode.p2j.util.*;
    import com.goldencode.p2j.util.logging.*;

    /**
     * This class implements the low level part of the protocol. It provides two
     * threads (Reader and Writer). Reader thread instantiates messages from the
     * stream and passes to the queue for the further processing. Writer thread
     * stores messages to stream. Also, <code>Protocol</code> is responsible for
     * notifying queue about I/O errors.
     * <p>
     * The reading and writing loops implement the proper callbacks for the
     * {@link StateSynchronizer} implementation of bidirectional state
     * synchronization.  This processing can be activated using the
     * <code>registerSynchronizer</code> method.
     * <p>
     * WARNING: the core loops of the reader and writer threads are controlled
     * by a variable which they read outside of a synchronized block. This means
     * that it is possible (depending on the JVM implementation) to get
     * inconsistent or corrupt results because it may be read in the middle of
     * being changed. At this time, we are not introducing a lock object since
     * this would cause potential contention to occur between the reader and
     * writer threads. This is something that needs to be tested to ensure its
     * impact is understood but it DOES need to be fixed.
     */
    class Protocol
    {
       /** Counter for measuring incoming traffic */
       private static final SimpleCounter INCOMING = SimpleCounter.getInstance(
          FwdServerJMX.Counter.NetworkReads
        );
       /** Counter for measuring outgoing traffic */
       private static final SimpleCounter OUTGOING = SimpleCounter.getInstance(
          FwdServerJMX.Counter.NetworkWrites
        );
       /** timer for measuring objToByteArray time */
       private static final NanoTimer TO_BYTE_ARRAY_TIMER = NanoTimer.getInstance(
          FwdServerJMX.TimeStat.objToByteArray
       );
       /** Logger. */
       private static final CentralLogger LOG = CentralLogger.get(Protocol.class.getName());

       /** Constant identifying a {@link Message} instance. */
       private static final int TYPE_MESSAGE = 0;

       /** Constant identifying a {@link SyncMessage} instance. */
       private static final int TYPE_SYNC_MESSAGE = 1;

       /** Thread group for the protocol threads. */
       private static ThreadGroup protoThreads = new ThreadGroup(
          Thread.currentThread().getThreadGroup(), "protocol");

       /** Flag indicating status for Reader/Writer threads. */
       private volatile boolean running = false;

       /** The queue representing the session with which protocol is working. */
       private Queue queue = null;

       /** Reader thread. */
       private Thread readerThread = null;

       /** Writer thread. */
       private Thread writerThread = null;

       /** Synchronizes access to the state synchronizer. */
       private Object slock = new Object();

       /** The object used to implement state synchronization if enabled. */
       private StateSynchronizer ssync = null;

       /** Synchronizes termination processing. */
       private Object tlock = new Object();

       /**
        * Constructs an instance using the provided queue.
        *
        * @param    queue
        *           Instance of the <code>Queue</code> to/from which messages
        *           are written/read.
        */
       Protocol(Queue queue)
       {
          this.queue = queue;
       }

       /**
        * Starts and Writer threads as daemon threads.  This occurs so
        * long as:
        * <ul>
        *    <li> The threads are not already started.
        *    <li> The queue was properly initialized.
        *    <li> The low-level network transport is not closed.
        * </ul>
        */
       synchronized void start()
       {
          if (running == true || queue == null || queue.isTransportClosed())
             return;

          running = true;

          Runnable reader = new Reader();
          Runnable writer = new Writer();

          readerThread = new Thread(protoThreads, reader, "Reader");
          readerThread.setDaemon(true);

          writerThread = new Thread(protoThreads, writer, "Writer");
          writerThread.setDaemon(true);

          readerThread.start();
          writerThread.start();
       }

       /**
        * Stops Reader and Writer threads.
        */
       synchronized void stop()
       {
          if (!running)
             return;

          running = false;
          readerThread.interrupt();
          writerThread.interrupt();

          queue.closeTransport();
          readerThread = null;
          writerThread = null;
       }

       /**
        * Force the reader and writer threads to have more descriptive names
        * to aid debugging. This is only meaningful on a router node.
        */
       synchronized void setNames()
       {
          SessionManager sessMgr = SessionManager.get();

          if (sessMgr.isRouter())
          {
             String rname = Utils.createThreadName("Reader");
             String wname = Utils.createThreadName("Writer");

             if (rname != null)
                readerThread.setName(rname);

             if (wname != null)
                writerThread.setName(wname);
          }
       }

       /**
        * Returns the operational state of this protocol driver.
        *
        * @return   <code>true</code> if the protocol is running (operational)
        *           and <code>false</code> if the session is closed.
        */
       public synchronized boolean isRunning()
       {
          return running;
       }

       /**
        * Sets the instance of the state synchronizer the reader and writer
        * threads should use, if and only if there is no other state synchronizer
        * already in use.  This "there can BE only ONE" approach can be changed
        * easily in the future but was implemented to ensure high performance
        * since each loop of the protocol must call the synchronization
        * primitives.
        *
        * @param    ssync
        *           The synchronizer to register or <code>null</code> if the
        *           currently registered synchronizer should be deregistered.
        */
       void registerSynchronizer(StateSynchronizer ssync)
       {
          synchronized (slock)
          {
             if (this.ssync == null || ssync == null)
             {
                this.ssync = ssync;
             }
          }
       }

       /**
        * Checks if there is a registered state synchronizer and if so, it is
        * queried to obtain the latest changes in state that must be sent to the
        * other side of the connection.  If any changes are found they are
        * packaged into the message and returned, otherwise the original message
        * is returned. If no state synchronizer is registered, the original
        * message is returned.
        * <p>
        * The state synchronization is allowed only for conversation mode or other processing which
        * is known to be synchronous.
        * <p>
        * For async replies or requests, no state services are used, as the state is something which
        * is required to be the same (and in sync) between both sides.  The very nature of async
        * messages may make the state to be inconsistent between the running threads.
        *
        * @param    outbound
        *           The message that was dequeued for transmission.
        *
        * @return   The updated message (with attached state) if there is a
        *           registered state synchronizer AND there are state changes
        *           to send.  Otherwise the original message will be returned.
        */
       Message attachChanges(Message outbound)
       {
          if (outbound.isAsyncReply() || outbound.isAsyncRequest())
          {
             // do not attach changes if the message is async - as this can get deadlocked with other
             // Conversation-thread processing.
             return outbound;
          }

          Externalizable state  = null;
          Message      result = outbound;

          synchronized (slock)
          {
             if (ssync != null)
             {
                state = ssync.getChanges();
             }
          }

          if (state != null)
          {
             result = new SyncMessage(outbound, state);
          }

          return result;
       }

       /**
        * If there is a registered state synchronizer and if there are any
        * attached state updates sent from the other side of the connection,
        * these updates are passed to the state synchronizer to be applied to
        * this node's state.
        *
        * @param    inbound
        *           The message that is about to be enqueued for processing.
        *
        * @return   An instance of the message without an attachment or the
        *           given message (unchanged) if there was no attachment.
        */
       Message applyChanges(Message inbound)
       {
          if (inbound instanceof SyncMessage)
          {
             SyncMessage  smsg  = (SyncMessage) inbound;
             Externalizable state = smsg.getState();

             synchronized (slock)
             {
                if (ssync != null)
                {
                   ssync.applyChanges(state);
                }
             }

             // remove the attachment (this copy constructor knows nothing of
             // the subclass so we will only get a simple message as a result)
             inbound = new Message(inbound);
          }

          return inbound;
       }

       /**
        * Stop the queue with SilentUnwindException checking.
        *
        * @param    result
        *           The managed exception.
        */
       private void stopQueue(Exception result)
       {
          try
          {
             if (running)
             {
                queue.stop(result, true);
             }
          }

          catch (SilentUnwindException ex)
          {
             LOG.log(Level.WARNING, "stopQueue", ex);
          }
       }

       /**
        * This helper class implements functionality of Reader thread.
        */
       private class Reader
       implements Runnable
       {
          /**
           * This method instantiates messages from the stream and places into
           * the queue for further processing.
           */
          public void run()
          {
             Exception result = null;

             try
             {

                while (running)
                {
                   // read the next message from the stream
                   byte[] data = queue.read();
                   if (data == null)
                   {
                      continue;
                   }
                   INCOMING.update(data.length);
                   // re-inflate the flattened object that came in as a byte[]
                   // (this is faster than without the intermediate flattened
                   // form)
                   Message msg = byteArrayToObj(data);

                   // we can already release data memory here
                   data = null;

                   // hand the message to the dispatcher for processing
                   queue.enqueueInbound(msg);
                }
             }

             catch (EOFException exc)
             {
                // this happens when the peer side terminates so this is NOT
                // a form of unexpected failure
             }

             catch (InterruptedException exc)
             {
                // an interrupted exception will be caused by the stop() method
                // so we consider this a "clean" way of exiting
             }

             catch (SocketException | SSLException exc)
             {
                // this happens when the socket is closed so this is NOT a form
                // of unexpected failure if the running flag was set to false
                if (running)
                {
                   // log that there was an unexpected failure
                   if (LOG.isLoggable(Level.FINE))
                   {
                      LOG.logp(Level.FINE,
                               "Protocol.Reader.run()",
                               "",
                               "Peer disconnected!",
                               exc);
                   }

                   result = exc;
                }
             }

             catch (Throwable thr)
             {
                // unexpected failure
                LOG.logp(Level.WARNING,
                         "Protocol.Reader.run()",
                         "",
                         "failure in reading loop",
                         thr);

                result = (thr instanceof Exception) ? (Exception) thr
                                                    : new RuntimeException(thr);
             }

             finally
             {
                // only one of the (reader and writer) threads should process
                // termination at any given time; the first one in gets to process
                // and the other will be blocked until the processing is complete;
                synchronized (tlock)
                {
                   // the second thread will find that the running flag is cleared
                   // so it won't need to and won't try to execute termination
                   // processing
                   stopQueue(result);
                }
             }
          }

          /**
           * Restore ("inflate") an object from an in-memory array of bytes.
           *
           * @param    obj
           *           The flattened object to restore.
           *
           * @return   The inflated object.
           * @throws   IOException
           *           on error
           * @throws   ClassNotFoundException
           *           if class specified in the binary stream was not found
           */
          private Message byteArrayToObj(byte[] obj)
          throws IOException,
                 ClassNotFoundException
          {
             Message msg = obj[0] == TYPE_MESSAGE ? new Message() : new SyncMessage();

             ByteArrayInputStream bais = new ByteArrayInputStream(obj, 1, obj.length);
             ObjectInputStream    ois  = new ObjectInputStream(bais);
             msg.readExternal(ois);

             return msg;
          }
       }

       /**
        * Custom version of the ByteArrayOutputStream with access to underlying buffer
        *
        */
       public class ByteArrayOutputStream2 extends ByteArrayOutputStream
       {
          /**
           * Constructor
           */
          public ByteArrayOutputStream2()
          {
             super();
          }
          /**
           * Constructor
           * @param size
           *       initial capacity
           */
          public ByteArrayOutputStream2(int size)
          {
             super(size);
         }

          /**
           * Returns the internal buffer of this ByteArrayOutputStream, without copying.
           * @return the internal buffer
           */
          public synchronized byte[] buf() {
              return this.buf;
          }
      }
       /**
        * This helper class implements functionality of Writer thread.
        */
       private class Writer
       implements Runnable
       {
          /**
           * Default size of buffer for object flattening
           */
          private static final int BUFFER_SIZE = 8192;

          /**
           * Pre-allocated helper buffers for object flattening
           */
          private ByteArrayOutputStream2 baos = new ByteArrayOutputStream2(BUFFER_SIZE);
          private ByteArrayOutputStream res = new ByteArrayOutputStream(BUFFER_SIZE);

          /**
           * Holder for data to be written out
           */
          private byte[] data;

          /**
           * Set new value of the <code>data</code> field (by be used in lambdas)
           * @param data
           *        new value of the field
           */
          private void setData(byte[] data)
          {
             this.data = data;
          }
          /**
           * This method obtains messages from the Outbound Queue and writes them
           * into output stream.
           */
          public void run()
          {
             Exception result = null;

             try
             {
                while (running)
                {
                   // dequeue a message
                   Message obj = queue.dequeueOutbound();

                   // flatten the object to a byte[] before writing it to the
                   // socket since this is significantly faster than going
                   // without the intermediate byte[]; note that we still use
                   // serialization for the final output since the low level
                   // Java approach handles reading an entire array as a single
                   // entity which is something we want to avoid doing ourselves

                   if (FwdServerJMX.JMX_DEBUG)
                   {
                      TO_BYTE_ARRAY_TIMER.timer(() -> setData(objToByteArray(obj)));
                   }
                   else
                   {
                      setData(objToByteArray(obj));
                   }
                   if (data != null)
                   {
                      OUTGOING.update(data.length);
                   }
                   queue.write(data);

                   // allow GC to collect memory allocated for data
                   data = null;
                }
             }

             catch (InterruptedException exc)
             {
                // an interrupted exception will be caused by the stop() method
                // so we consider this a "clean" way of exiting
             }

             catch (SocketException exc)
             {
                // this happens when the socket is closed so this is NOT a form
                // of unexpected failure if the running flag was set to false
                if (running)
                {
                   // log that there was an unexpected failure
                   if (LOG.isLoggable(Level.FINE))
                   {
                      LOG.logp(Level.FINE,
                               "Protocol.Writer.run()",
                               "",
                               "Peer disconnected!",
                               exc);
                   }

                   result = exc;
                }
             }

             catch (Throwable thr)
             {
                // unexpected failure
                LOG.logp(Level.WARNING,
                         "Protocol.Writer.run()",
                         "",
                         "failure in writing loop",
                         thr);

                result = (thr instanceof Exception) ? (Exception) thr
                                                    : new RuntimeException(thr);
             }

             finally
             {
                // only one of the (reader and writer) threads should process
                // termination at any given time; the first one in gets to process
                // and the other will be blocked until the processing is complete;
                synchronized (tlock)
                {
                   // the second thread will find that the running flag is cleared
                   // so it won't need to and won't try to execute termination
                   // processing
                   stopQueue(result);
                }
             }
          }

          /**
           * Serialize the given object into an in-memory array of bytes.
           *
           * @param    obj
           *           The graph of objects to be converted.
           *
           * @return   The "flattened" version.
           * @throws IOException
           *         on error
           */
          private byte[] objToByteArray(Message obj)
          throws IOException
          {
             baos.reset();
             ObjectOutputStream    oos  = new ObjectOutputStream(baos);
             obj.writeExternal(oos);
             oos.flush();

             res.reset();
             res.write(obj.getClass() == Message.class ? TYPE_MESSAGE : TYPE_SYNC_MESSAGE);
             res.write(baos.buf(), 0, baos.size());

             // in case baos grows too big -> reallocate it
             if (baos.size() > BUFFER_SIZE)
             {
                baos = new ByteArrayOutputStream2(BUFFER_SIZE);
             }

             byte[] result = res.toByteArray();

             // in case res grows too big -> reallocate it
             if (res.size() > BUFFER_SIZE)
             {
                res = new ByteArrayOutputStream(BUFFER_SIZE);
             }

             return result;
          }
       }
    }