context: /*
      Copyright (c) 2009-2011 250bpm s.r.o.
      Copyright (c) 2007-2009 iMatix Corporation
      Copyright (c) 2011 VMware, Inc.
      Copyright (c) 2007-2015 Other contributors as noted in the AUTHORS file
  
      This file is part of 0MQ.
  
      0MQ is free software; you can redistribute it and/or modify it under
      the terms of the GNU Lesser General Public License as published by
      the Free Software Foundation; either version 3 of the License, or
      (at your option) any later version.
  
      0MQ 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 Lesser General Public License for more details.
  
      You should have received a copy of the GNU Lesser General Public License
      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
  
  using System.Diagnostics;
  using JetBrains.Annotations;
  
  namespace NetMQ.Core
  {
      /// <summary>
      /// A pipe is a ZObject with ..
      /// </summary>
      /// <remarks>
      /// Note that pipe can be stored in three different arrays.
      /// The array of inbound pipes (1), the array of outbound pipes (2) and
      /// the generic array of pipes to deallocate (3).
      /// </remarks>
      internal class Pipe : ZObject
      {
          public interface IPipeEvents
          {
              /// <summary>
              /// Indicate that the given pipe is now ready for reading.
              /// Pipe calls this on it's sink in response to ProcessActivateRead.
              /// When called upon an instance of SocketBase, this simply calls XReadActivated.
              /// </summary>
              /// <param name="pipe">the pipe to indicate is ready for reading</param>
              void ReadActivated([NotNull] Pipe pipe);
  
              void WriteActivated([NotNull] Pipe pipe);
              void Hiccuped([NotNull] Pipe pipe);
  
              /// <summary>
              /// This gets called by ProcessPipeTermAck or XTerminated to respond to the termination of the given pipe.
              /// </summary>
              /// <param name="pipe">the pipe that was terminated</param>
              void Terminated([NotNull] Pipe pipe);
          }
  
          /// <summary>
          /// The underlying pipe for reading from.
          /// </summary>
          private YPipe<Msg> m_inboundPipe;
  
          /// <summary>
          /// The underlying pipe for writing to.
          /// </summary>
          private YPipe<Msg> m_outboundPipe;
  
          /// <summary>
          /// This indicates whether this pipe can be read from.
          /// </summary>
          private bool m_inActive;
  
          /// <summary>
          /// This indicates whether this pipe be written to.
          /// </summary>
          private bool m_outActive;
  
          /// <summary>
          /// High watermark for the outbound pipe.
          /// </summary>
          private readonly int m_highWatermark;
  
          /// <summary>
          /// Low watermark for the inbound pipe.
          /// </summary>
          private readonly int m_lowWatermark;
  
          /// <summary>
          /// Number of messages read so far.
          /// </summary>
          private long m_numberOfMessagesRead;
  
          /// <summary>
          /// Number of messages written so far.
          /// </summary>
          private long m_numberOfMessagesWritten;
  
          /// <summary>
          /// Last received peer's msgs_read. The actual number in the peer
          /// can be higher at the moment.
          /// </summary>
          private long m_peersMsgsRead;
  
          /// <summary>
          /// The pipe object on the other side of the pipe-pair.
          /// </summary>
          private Pipe m_peer;
  
          /// <summary>
          /// Sink to send events to.
          /// </summary>
          private IPipeEvents m_sink;
  
          /// <summary>
          /// Specifies the state of the pipe endpoint.
          /// </summary>
          private enum State
          {
              /// <summary> Active is common state before any termination begins. </summary>
              Active,
              /// <summary> Delimited means that delimiter was read from pipe before term command was received. </summary>
              Delimited,
              /// <summary> Pending means that term command was already received from the peer but there are still pending messages to read. </summary>
              Pending,
              /// <summary> Terminating means that all pending messages were already read and all we are waiting for is ack from the peer. </summary>
              Terminating,
              /// <summary> Terminated means that 'terminate' was explicitly called by the user. </summary>
              Terminated,
              /// <summary> Double_terminated means that user called 'terminate' and then we've got term command from the peer as well. </summary>
              DoubleTerminated
          }
  
          private State m_state;
  
          /// <summary>
          /// If <c>true</c>, we receive all the pending inbound messages before terminating. 
          /// If <c>false</c>, we terminate immediately when the peer asks us to.
          /// </summary>
          private bool m_delay;
  
          /// <summary>
          /// Identity of the writer. Used uniquely by the reader side.
          /// </summary>
          private readonly ZObject m_parent;
  
          /// <summary>
          /// Create a new Pipe object with the given parent, and inbound and outbound YPipes.
          /// </summary>
          /// <remarks>
          /// Constructor is private as pipe can only be created using <see cref="PipePair"/> method.
          /// </remarks>
          private Pipe(
              [NotNull] ZObject parent, [NotNull] YPipe<Msg> inboundPipe, [NotNull] YPipe<Msg> outboundPipe,
              int inHighWatermark, int outHighWatermark, int predefinedLowWatermark, bool delay)
              : base(parent)
          {
              m_parent = parent;
              m_inboundPipe = inboundPipe;
              m_outboundPipe = outboundPipe;
              m_inActive = true;
              m_outActive = true;
              m_highWatermark = outHighWatermark;
              m_lowWatermark = ComputeLowWatermark(inHighWatermark, predefinedLowWatermark);
              m_numberOfMessagesRead = 0;
              m_numberOfMessagesWritten = 0;
              m_peersMsgsRead = 0;
              m_peer = null;
              m_sink = null;
              m_state = State.Active;
              m_delay = delay;
          }
  
          /// <summary>
          /// Create a pipe pair for bi-directional transfer of messages.
          /// </summary>
          /// <param name="parents">The parents.</param>
          /// <param name="highWaterMarks">First HWM is for messages passed from first pipe to the second pipe.
          /// Second HWM is for messages passed from second pipe to the first pipe.</param>
          /// <param name="lowWaterMarks">First LWM is for messages passed from first pipe to the second pipe.
          /// Second LWM is for messages passed from second pipe to the first pipe.</param>
          /// <param name="delays">Delay specifies how the pipe behaves when the peer terminates. If true
          /// pipe receives all the pending messages before terminating, otherwise it
          /// terminates straight away.</param>
          /// <returns>A pipe pair for bi-directional transfer of messages. </returns>
          [NotNull]
          public static Pipe[] PipePair([NotNull] ZObject[] parents, [NotNull] int[] highWaterMarks, [NotNull] int[] lowWaterMarks, [NotNull] bool[] delays)
          {
              // Creates two pipe objects. These objects are connected by two ypipes,
              // each to pass messages in one direction.
  
              var upipe1 = new YPipe<Msg>(Config.MessagePipeGranularity, "upipe1");
              var upipe2 = new YPipe<Msg>(Config.MessagePipeGranularity, "upipe2");
  
              var pipes = new[]
              {
                  new Pipe(parents[0], upipe1, upipe2, highWaterMarks[1], highWaterMarks[0], lowWaterMarks[1], delays[0]),
                  new Pipe(parents[1], upipe2, upipe1, highWaterMarks[0], highWaterMarks[1], lowWaterMarks[0], delays[1])
              };
  
              pipes[0].SetPeer(pipes[1]);
              pipes[1].SetPeer(pipes[0]);
  
              return pipes;
          }
  
          public bool Active => m_state == State.Active;
  
          /// <summary>
          /// <see cref="PipePair"/> uses this function to let us know about the peer pipe object.
          /// </summary>
          /// <param name="peer">The peer to be assigned.</param>
          private void SetPeer([NotNull] Pipe peer)
          {
              // Peer can be set once only.
              Debug.Assert(peer != null);
              m_peer = peer;
          }
  
          /// <summary>
          /// Specifies the object to send events to.
          /// </summary>
          /// <param name="sink"> The receiver of the events. </param>
          public void SetEventSink([NotNull] IPipeEvents sink)
          {
              Debug.Assert(m_sink == null);
              m_sink = sink;
          }
  
          /// <summary>
          /// Get or set the byte-array that comprises the identity of this Pipe.
          /// </summary>
          [CanBeNull]
          public byte[] Identity { get; set; }
  
          /// <summary>
          /// Checks if there is at least one message to read in the pipe.
          /// </summary>
          /// <returns> Returns <c>true</c> if there is at least one message to read in the pipe; <c>false</c> otherwise. </returns>
          public bool CheckRead()
          {
              if (!m_inActive || (m_state != State.Active && m_state != State.Pending))
                  return false;
  
              // Check if there's an item in the pipe.
              if (!m_inboundPipe.CheckRead())
              {
                  m_inActive = false;
                  return false;
              }
  
              // If the next item in the pipe is message delimiter,
              // initiate termination process.
              if (m_inboundPipe.Probe().IsDelimiter)
              {
                  var msg = new Msg();
  
gt:                 bool ok = m_inboundPipe.TryRead(out msg);