TCP Socket Off-the-shelf

• Download source - 26.6 KB
• Download demo - 18.3 KB

Introduction

In modern software systems, data transfer is usually carried out using serialized objects of types well-known to both the client and server. However, in many cases, continuous streams of bytes and raw TCP sockets are still used (e.g., due to performance or compatibility considerations). I recently came across such a communication for transfer of financial information, medical data, and “soft” sensors output. Although TCP sockets are well documented, their usage requires writing of tedious and sometimes tricky code. This article presents a TCP socket wrapper allowing the developer to use it as a “black box” and saving him/her from dealing with the implementation details.

The wrapper provides the following features:

• implementation of TCP client (connection initiator) and server (connection acceptor)
• synchronous (blocking) and asynchronous (non-blocking) data sending
• processing received data in a separate dedicated thread for each socket
• processing of continuous data stream
• notification on significant events, errors, and exceptions
• slow receiver identification by sender
• possibility to make client wait until server starts
• automatic configurable reconnection attempts when connection lost

Most of the above features will be discussed in detail in the article.

Code Description

The wrapper itself is placed in the assembly IL.TcpCommunicationLib.dll. It contains a base class TcpChannel implementing most of the functionality, and two derived classes, namely, TcpServer and TcpClient. The assembly IL.WorkerThreadLib.dll provides a supporting WorkerThread class for multithreading (this type may be used independently, apart from communication purposes for any cyclic processing in dedicated threads). To establish communication, the developer has to provide a local IP address for the server and the client (alternatively it can be obtained automatically), callback methods for received data processing and notifications (optionally), and

• for server: call the static method TcpServer.StartAcceptSubscribersOnPort() to start listening for incoming connections on a given port
• for client: create an object of TcpClient type and call its method Connect() to initiate connection to the server

Communication using the wrapper classes is performed as follows. Both the Server and Client provide / automatically obtain their respective local IP addresses. The Server starts to listen for incoming connections on a specified port by calling the static method TcpServer.StartAcceptSubscribersOnPort(). The Client calls the method Connect() providing the Server's IP address and port that the Server is listening on. The Server accepts the Client's call and actually establishes connection with the Client automatically allocating for this connection some port different from the one listening on. Upon connection established, both Server and Client internally call the method Receive() to receive arrays of bytes sent by the other side. TcpChannel.Receive() receives data from the socket synchronously. This is done to ensure strictly sequential bytes processing and to avoid too many threads from the framework thread pool being involved. The Receive() method provides continuous listening for incoming byte stream and places the received bytes in a thread safe queue for processing. The received bytes are processed in a dedicated thread of a WorkerThread type object. Receive() provides callback delegates for both reading and processing threads. The processing thread callback method calls in its turn the onReceived event supplied by the user in the TcpClient (and down to TcpChannel) constructor for the Client, and in the callback within AcceptBegin() for the Server.

During data exchange, the onReceived event of type EventHandler<TcpChannelReceivedEventArgs> is called periodically by the processing thread to parse the received bytes. The event handler should be implemented by the user. Its first parameter is of type TcpChannel and the second is of type TcpChannelReceivedEventArgs containing a chunk of received bytes. Since onReceived is always called from the same processing thread, its handler is thread safe.

Continuously received bytes should be parsed to obtain required data, usually in the form of objects of well-known types. The incoming stream may be split into certain records (objects) by delimiters or using fixed fields length. A given fragment of incoming string may contain an incomplete record at the start and at the end. To deal with such situations, the TcpChannel class provides a property UnparsedBytes of type byte[] to save bytes between the last complete record and the end of the currently received chunk of data. These bytes (if any) will be placed before the next chunk of received data in the TcpChannelReceivedEventArgs.BtsReceived property to ensure that this property always starts with a new record. It is the responsibility of the user provided onReceived event handler to fill the TcpChannel.UnparsedBytes property with the leading bytes of the last [incomplete] record at the end of the current chunk parsing. It is preferred to keep the unparsed bytes in the TcpChannel object itself over keeping them in a caller object because the caller object may contain several TcpChannel objects forcing the usage of an additional [synchronized] dictionary.

Events and Diagnostics

Strictly speaking, the handler for the onReceived event is the only compulsory callback that should be implemented by the TcpChannel user. But the socket wrapper types provide a set of notifications about a variety of their internal events providing the caller with relevant data. These notifications are available as events of type EventHandler<TcpChannelNotifyEventArgs>. They may be used for state switch, diagnostics, and logging. In addition to TcpChannel notification events, TcpServer may use the static onServerNotifies to report events happened in TcpServer's static methods before a connection is established. The onServerNotifies event handler implemented by the caller is the fourth optional parameter of the TcpServer.StartAcceptSubscribersOnPort() static method. TcpServer may provide a more special event onInitConnectionToServer. Its delegate is called by the Server when the Server has accepted an incoming connection request from the Client, created an appropriate object of TcpServer (the first "sender" argument in the event call), and allotted a socket for this connection. In its handler, Server may send some connection acknowledgement message to the Client. onInitConnectionToServer constitutes the third optional parameter of the TcpServer.StartAcceptSubscribersOnPort() static method.

For the sender side, it is often important to identify the situation of "slow receiver", that is when the time required for the socket to perform a send operation exceeds the parameter SendTimeout of the socket. This can happen, e.g., due to not wide enough network bandwidth or when the receiver side gets data too slow. The type TcpChannel diagnoses this situation for both synchronous (by analyzing the exception brought by the onSyncSendException event) and asynchronous (onAsyncSendSendingTimeoutExceeded event) sending modes.

Code Sample

The small application SampleApp.exe illustrates the usage of the socket wrapper types. Depending on its arguments, the application acts either as a Server or as a Client. The following code fragment presents part of its Main() method.

// Set local host for all channels (both server and client) in the machine
TcpChannel.LocalHost = localHost;
onReceived = new TcpChannelEventHandler<tcpchannelreceivedeventargs>((tcpChannelSender, e) =>
    {
        if (e.AreBytesAvailable)
        {
            // Some processing of received bytes
            if (tcpChannelSender != null)
                tcpChannelSender.Send(/* bytes or string to send */));
        }
    });

switch (role)
{
    case Role.Server:
        onInitConnectionToServer = new TcpChannelEventHandler<EventArgs>((tcpServerSender, e) =>
            {
                SetEventHandlers(tcpServerSender);
                tcpServerSender.Send(/* bytes or string to send */);
            });
        onServerNotifies = new TcpChannelEventHandler<TcpChannelNotifyEventArgs>(
            (tcpServerSender, e) =>
            {
                // Some notification processing, e.g. logging
            });
        TcpServer.StartAcceptSubscribersOnPort(localPort, onReceived,
                            onInitConnectionToServer, onServerNotifies);
        isListening = true;
        break;

    case Role.Client:
        TcpClient tcpClient = new TcpClient(onReceived, 
                                            localPort.ToString(),// id
                                            15,                  // receiveTimeoutInSec
                                             9,                  // reconnectionAttempts
                                            10);               // delayBetweenReconnectionAttemptsInSec  
        SetEventHandlers(tcpClient);                                            
        tcpClient.Connect(remoteHost, remotePort);
        break;
}

Let's discuss the above code. Both the Server and Client caller applications need to implement a handler for the onReceived event. The Server also implements handlers for the onInitConnectionToServer and onServerNotifies events. The Server calls the static method TcpServer.StartAcceptSubscribersOnPort() to start listening on localPort for incoming connection requests from Clients. When a connection has been established, the onInitConnectionToServer event is called with the newly created TcpServer object as sender (the first argument). The Client first creates a TcpClient object by calling the TcpClient public constructor. To provide a certain degree of flexibility, the static method TcpServer.StartAcceptSubscribersOnPort() and the constructor of the TcpClient type have several arguments, most of which however have default values. The table below provides information about the arguments:

The method SetEventHandlers() assigns handlers to notification events of TcpChannel and TcpClient.

static void SetEventHandlers(TcpChannel tcpChannel)
{
    if (tcpChannel == null)
        return;

    tcpChannel.onSocketNullOrNotConnected += ((tcpChannelSender, e) => 
        { 
            // Event handler    
        });

    // Other event handles assignment for TcpChannel

    TcpClient tcpClient = tcpChannel as TcpClient;
    if (tcpClient != null)
    {
        tcpClient.onSocketConnectionFailed += ((tcpClientSender, e) =>
            {
                // Event handler    
            });

        // Other event handles assignment for TcpClient
    }
}

In the case of the Server, this method is called by the onInitConnectionToServer event handler, and in the case of the Client, the method immediately follows the TcpClient constructor.

After the call to the SetEventHandlers() method the Client calls the method Connect() providing the address and port of the Server as arguments. If the Server is listening for an incoming connection and is ready to accept one, then in the Server synchronization event, evAccept is set in the callback defined in the method tcpListener.BeginAcceptSocket(), an object of TcpServer type is created, a connection with the Client established, and the event onInitConnectionToServer is called. In case the Server is not listening at the moment for an incoming connection, it is still possible for the Client to keep sending a connection request periodically until the Server is ready to accept it. To enable this feature, a configurable reconnection mechanism is supported by the TcpClient and TcpChannel types. This mechanism is configured with three arguments of the TcpClient constructor, namely, receiveTimeoutInSec, reconnectionAttempts, and delayBetweenReconnectionAttemptsInSec. Their description and default values are given in the table above. By default, the reconnection mechanism is disabled (reconnectionAttempts is equal to 0). But with appropriate configuration (like, e. g., in the code fragments above), reconnection attempts will be periodically carried out including the initial connection to the Server. Reconnection is exclusively a Client feature, as any connection initiation. As soon as a connection between the Client and Server is established, both parties can send (synchronously or asynchronously) arrays of bytes, receive, and process received bytes with the onReceived event handler.

To run the demo, command files RunServer.cmd, RunClient1.cmd, and RunClient2.cmd containing SampleApp.exe with appropriate arguments should be started. If Clients are started before the Server, then a connection will happen within delayBetweenReconnectionAttemptsInSec (in the sample, this is 10 sec) after the Server's start. In order to demonstrate reconnection in action, the Server demo sample has a timer causing Server's cyclic work: the Server works for a minute and then remains idle for another minute. After the Server resumes its work, Clients get connected to the Server again within delayBetweenReconnectionAttemptsInSec. During the Server's idle time, Clients keep trying to connect to the Server according to their reconnection configuration.

Conclusion

This article presents a small library that simplifies continuous data exchange through TCP sockets. Very little user's code is required. Types of the library are equipped with a set of notification events for diagnostics and logging purposes. A socket receives data synchronously with their processing in another dedicated thread.

Thanks

Many thanks to a great professional and a friend of mine, Michael Molotsky, for a very useful discussion on the topic of this article.