FCC/Extension/GlobalPumpFdc/GlobalFccClient/MessageRouterConnection.cs

using System;
using System.Text;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Threading;

namespace MessageRouter
{
    public delegate void OnMessageRouterConnectDelegate(MessageRouterConnection client, bool reconnect);

    public delegate void OnMessageRouterDisconnectDelegate(MessageRouterConnection client, Exception e);

    public delegate void OnMessageRouterPacketReceivedDelegate(MessageRouterConnection client, MessageRouterPacket p);

    /// <summary>
    /// Abstraction of Fusion Message Router Packet
    /// <remoarks>
    /// Fusion Message Router Message Format:
    /// <len>|<crypt format>|<message version>|<user id>|<message type>|<event type>|<destination>|<origin>|<parameters…>|^
    /// <len>: 5 bytes fixed length field containing the length of the message from the <message version> till the ^ included.
    /// <crypt format>: 1 byte fixed length field that describes the crypt format of the message.  Fixed value '5'. This means no encryption.
    /// </remarks>
    /// </summary>
    public class MessageRouterPacket
    {
        public static int MAX_PACKET_LENGTH = 1024 * 1024;
        public static int LEN_LENGTH = 5;
        public static int CRYPT_FORMAT_LENGTH = 1;
        public static int PIPE_LENGTH = 1;

        public readonly char cryptFormat;
        public readonly byte[] buf;

        public MessageRouterPacket(char cryptFormat, byte[] buf)
        {
            this.cryptFormat = cryptFormat;
            this.buf = buf;
        }
    }

    public class MessageRouterConnection : IDisposable
    {
        /// <summary> Remote end point </summary>
        private IPEndPoint remoteEp;

        /// <summary> Port of local endpoint </summary>
        private int localPort;

        /// <summary> Connection delay </summary>
        private readonly int delay;

        /// <summary> The default delay between atttempts to connect (in ms). </summary>
        private const int DEFAULT_DELAY = 5000;

        #region Property accessor
        public IPEndPoint RemoteEp
        {
            get { return remoteEp; }
            set { remoteEp = value; }
        }

        public int LocalPort
        {
            get { return localPort; }
        }
        #endregion


        /// <summary>
        /// Constructs the ipc client with default write queue length and default delay.
        /// </summary>
        /// <param name="remoteEp">The remote endpoint to use. May be specified as
        /// null and set later using assignment to RemoteEp.</param>
        public MessageRouterConnection(IPEndPoint remoteEp)
            : this(remoteEp, DEFAULT_DELAY)
        {
        }

        /// <summary>
        /// Construct IPC client
        /// </summary>
        /// <param name="remoteEp">The remote endpoint to use.</param>
        /// <param name="delay">Delay in milliseconds between attempts to connect.
        /// specify 0 to not delay any more than necessary between attempts.
        /// </param>
        public MessageRouterConnection(IPEndPoint remoteEp, int delay)
        {
            if (delay < 0)
                throw new ArgumentException("Invalid delay");

            this.remoteEp = remoteEp;
            this.delay = delay;
            this.localPort = 0;
        }

        /// <summary>
        /// Helper to convert object state to readable string
        /// </summary>
        /// <returns>String with object properties</returns>
        public override String ToString()
        {
            return "MessageRouterConnection(" + remoteEp + ", " + state + ")";
        }

        #region Socket event delegates
        /// <summary> Delegate to inform when a connection is established. </summary>
        public OnMessageRouterConnectDelegate OnConnect;

        /// <summary> Delegate to inform when a connection is closed. </summary>
        public OnMessageRouterDisconnectDelegate OnDisconnect;

        /// <summary> Delegate to inform when a packet is received. </summary>
        public OnMessageRouterPacketReceivedDelegate OnPacketReceived;
        #endregion

        #region Packet I/O
        /// <summary>
        /// Read packet from memory stream
        /// </summary>
        /// <param name="s">Memory stream</param>
        /// <returns>Packet from memory stream</returns>
        public MessageRouterPacket ReadPacket(Stream s)
        {
            // Read message length, 5 bytes
            int len = ReadLen(s);

            if (len == 0)
                throw new IOException("Message router packet len = 0");

            if (len > MessageRouterPacket.MAX_PACKET_LENGTH)
                throw new IOException("Message router packet len > Max packet size");

            // Read Pipe char
            if (!ReadPipe(s))
                throw new IOException("Malformed router message");

            // Read crypto format
            char cryptFormat = ReadCryptFormat(s);

            // Read Pipe char
            if (!ReadPipe(s))
                throw new IOException("Malformed router message");

            byte[] buf = new byte[len];
            ReadBytes(s, buf);

            return new MessageRouterPacket(cryptFormat, buf);
        }

        private int ReadLen(Stream s)
        {
            byte[] buf = new byte[MessageRouterPacket.LEN_LENGTH];
            ReadBytes(s, buf);

            Encoding encoding = new ASCIIEncoding();
            //Console.WriteLine("READ LEN: " + encoding.GetString(buf));
            String str = encoding.GetString(buf).TrimStart('0');
            return Convert.ToInt32(str);            
        }

        private char ReadCryptFormat(Stream s)
        {
            byte[] buf = new byte[MessageRouterPacket.CRYPT_FORMAT_LENGTH];
            ReadBytes(s, buf);

            char cf = (char)buf[0];

            //Console.WriteLine("READ CRYPT: " + cf);

            return cf;
        }

        private bool ReadPipe(Stream s)
        {
            byte[] buf = new byte[MessageRouterPacket.PIPE_LENGTH];
            ReadBytes(s, buf);
            if (buf[0] != 0x7C)
                return false;

            return true;
        }

        /// <summary>
        /// Read byte array from socket stream
        /// </summary>
        /// <param name="s">Socket data stream</param>
        /// <param name="buf">Data buffer</param>
        private void ReadBytes(Stream s, byte[] buf)
        {
            if (s == null)
                throw new IOException("Socket closed");

            int n = buf.Length;
            int i = 0;
            while (i < n)
            {
                int k = s.Read(buf, i, n - i);
                if (k == 0)
                    throw new IOException("Socket eof");
                i += k;
            }
        }

        private void WriteLen(Stream s, int len)
        {
            byte[] buf = new byte[MessageRouterPacket.LEN_LENGTH];

            String str = len.ToString();
            int padding = MessageRouterPacket.LEN_LENGTH - str.Length;
            for (int i = 0; i < padding; i++)
                str = "0" + str;

            System.Text.ASCIIEncoding encoding = new System.Text.ASCIIEncoding();

            WriteBytes(s, encoding.GetBytes(str));            
        }

        private void WriteCryptFormat(Stream s, char cryptFormat)
        {
            byte[] buf = new byte[MessageRouterPacket.CRYPT_FORMAT_LENGTH];
            buf[0] = (byte)cryptFormat;
            WriteBytes(s, buf);
        }

        private void WritePipe(Stream s)
        {
            byte[] pipe = new Byte[MessageRouterPacket.PIPE_LENGTH];
            pipe[0] = 0x7C;
            WriteBytes(s, pipe);
        }

        // Write byte array into socket stream
        private void WriteBytes(Stream s, byte[] buf)
        {
            if (s == null)
                throw new IOException("Socket closed");

            s.Write(buf, 0, buf.Length);
        }

        /// <summary>
        /// Write packet into socket stream
        /// </summary>
        /// <param name="s">Data stream</param>
        /// <param name="p">Data packet</param>
        private void WritePacket(Stream s, MessageRouterPacket p)
        {
            WriteLen(s, p.buf == null ? 0 : p.buf.Length);

            WritePipe(s);

            WriteCryptFormat(s, p.cryptFormat);

            WritePipe(s);

            WriteBytes(s, p.buf);

            s.Flush();
        }
        #endregion

        #region Socket state management
        enum State { CLOSED, OPENING, OPENED };

        /// <summary> Socket state </summary>
        private State _state;

        /// <summary> Socket state accessor </summary>
        private State state
        {
            get
            {
                return _state;
            }

            set
            {
                lock (this)
                {
                    _state = value;
                    Monitor.PulseAll(this);
                }
            }
        }
        #endregion

        #region Connection Management
        /// <summary> TCP client object for this IPC client </summary>
        private TcpClient tcpClient;

        /// <summary> Input stream for this TCP client </summary>
        private Stream inputStream;

        /// <summary> Output stream for this TCP client </summary>
        private Stream outputStream;

        /// <summary> Flag to indicate TCP connection is reconnected </summary>
        private bool reconnect;

        /// <summary>
        /// Validate socket state before opening
        /// </summary>
        private void PrepareToOpen()
        {
            if (state != State.CLOSED)
                throw new InvalidOperationException("Socket already open");

            if (remoteEp == null)
                throw new InvalidOperationException("Invalid remote end point");

            if (OnPacketReceived == null)
                throw new InvalidOperationException("Missing deletegate for packet received event");

            state = State.OPENING;
        }

        /// <summary>
        /// Open a TCP connection to IPC server
        /// </summary>
        private void OpenConnection()
        {
            lock (this)
            {
                if (state != State.OPENING)
                    throw new Exception("Socket state not opening - " + state);

                tcpClient = new TcpClient();
                try
                {
                    tcpClient.Connect(remoteEp);    // connect to IPC server
                    tcpClient.NoDelay = true;       // disable nagle's algorithm

                    NetworkStream networkStream = tcpClient.GetStream();
                    inputStream = new BufferedStream(networkStream);
                    outputStream = new BufferedStream(networkStream);

                    state = State.OPENED;

                    IPEndPoint localEndPoint = tcpClient.Client.LocalEndPoint as IPEndPoint;
                    localPort = localEndPoint.Port;

                    if (OnConnect != null)
                        OnConnect(this, reconnect); // notify connection is established

                    reconnect = true;               // show that we've connected at least once
                }
                catch (Exception e)
                {
                    CloseConnection(e);
                    throw e;
                }
            }
        }

        /// <summary>
        /// Read data from input stream
        /// </summary>
        private void ReadConnection()
        {
            Stream xis = inputStream;
            if (xis == null)
                return;

            while (state == State.OPENED)
            {
                MessageRouterPacket p = ReadPacket(xis);
                if (OnPacketReceived != null)
                    OnPacketReceived(this, p);
            }
        }

        /// <summary>
        /// Close socket connection to IPC server
        /// </summary>
        /// <param name="e">Exception info</param>
        private void CloseConnection(Exception e)
        {
            lock (this)
            {
                if (tcpClient != null)
                {
                    if (inputStream != null)
                    {
                        try
                        {
                            inputStream.Close();
                        }
                        catch { }
                        inputStream = null;
                    }

                    if (outputStream != null)
                    {
                        try
                        {
                            outputStream.Close();
                        }
                        catch { }
                        outputStream = null;
                    }

                    try
                    {
                        tcpClient.Close();
                    }
                    catch { }
                    tcpClient = null;

                    localPort = 0;

                    if (state == State.OPENED)
                        state = State.OPENING;      // force reconnect

                    if (OnDisconnect != null)
                        OnDisconnect(this, e);
                }
            }
        }

        /// <summary>
        /// Permanently disposes of this object.
        /// </summary>
        public void Dispose()
        {
            Close(1); // close and don't wait!
        }
        #endregion

        #region Asynchronous Session Management
        /// <summary>
        /// Start socket connection thread
        /// </summary>
        public void Start()
        {
            lock (this)
            {
                PrepareToOpen();
                thread = StartThread(new ThreadStart(Run), "Run");
            }
        }

        /// <summary>
        /// Worker thead for connection management
        /// </summary>
        private void Run()
        {
            while (state != State.CLOSED)
            {
                OpenConnectionX();
                try
                {
                    ReadConnection();
                    CloseConnection(null);
                }
                catch (ThreadAbortException)
                {  // nothing
                }
                catch (Exception e)
                {
                    CloseConnection(e);
                }
            }
        }

        /// <summary>
        /// Open a TCP connection to IPC server until successful
        /// </summary>
        private void OpenConnectionX()
        {
            while (state == State.OPENING)
            {
                lock (this)
                {
                    try
                    {
                        OpenConnection();
                        return;
                    }
                    catch (SocketException)
                    {
                        if (state == State.OPENING)
                            Monitor.Wait(this, delay);
                    }
                }
            }
        }

        /// <summary>
        /// Close a socket connection without wait
        /// </summary>
        public void Close()
        {
            Close(0);
        }

        /// <summary>
        /// Close a socket connection with timeout
        /// </summary>
        /// <param name="timeout">Time to wait for socket to close</param>
        public void Close(int timeout)
        {
            if (timeout < 0)
                throw new ArgumentException("Invalid timeout value");

            lock (this)
            {
                if (state != State.CLOSED)
                {
                    state = State.CLOSED;

                    CloseConnection(new Exception("Socket closed"));
                }
            }

            if (thread != Thread.CurrentThread)
            {
                StopThread(thread, timeout);
                thread = null;
            }
        }
        #endregion

        #region Synchronous Session Management
        /// <summary>
        /// Open socket connection to IPC server synchronously
        /// </summary>
        /// <returns>True if connection established</returns>
        public bool Open()
        {
            lock (this)
            {
                try
                {
                    PrepareToOpen();
                    OpenConnection();
                    thread = StartThread(new ThreadStart(SyncReader), "SyncReader");
                    return true;
                }
                catch (SocketException e)
                {
                    state = State.CLOSED;
                    CloseConnection(e);
                    return false;
                }
            }
        }

        /// <summary>
        /// Worker thread logic to read data from input stream
        /// </summary>
        private void SyncReader()
        {
            try
            {
                ReadConnection();
                FinishSyncReader(null);
            }
            catch (Exception e)
            {
                FinishSyncReader(e);
            }
        }

        /// <summary>
        /// Close socket connection to IPC server
        /// </summary>
        /// <param name="e"></param>
        private void FinishSyncReader(Exception e)
        {
            lock (this)
            {
                if (state != State.CLOSED)
                {
                    state = State.CLOSED;
                    CloseConnection(e);
                    thread = null;
                }
            }
        }
        #endregion

        #region Thread Management
        /// <summary> The worker thead to create </summary>
        private Thread thread;

        /// <summary>
        /// Start worker thread
        /// </summary>
        /// <param name="start">Main function for this worker thread</param>
        /// <param name="what">Thread description</param>
        /// <returns>The created thread</returns>
        private Thread StartThread(ThreadStart start, string what)
        {
            Thread t = new Thread(start);
            t.IsBackground = true;
            t.Name = ToString() + " " + what;
            t.Start();
            return t;
        }

        /// <summary>
        /// Stop the worker thread
        /// </summary>
        /// <param name="t">Thread object to stop</param>
        /// <param name="timeout">Time to wait</param>
        private void StopThread(Thread t, int timeout)
        {
            if (t == null)
                return;

            if (!t.Join(timeout))
                t.Abort();
        }
        #endregion

        #region Outgoing Data
        public bool Write(char cryptFormat, byte[] data)
        {
            try
            {
                MessageRouterPacket p = new MessageRouterPacket(cryptFormat, data);

                lock (writeSync)
                {
                    WritePacket(outputStream, p);
                }
            }
            catch (Exception e)
            {
                CloseConnection(e);
                return false;
            }

            return true;
        }

        private Object writeSync = new Object();

        #endregion
    }
}