23
\$\begingroup\$

The following code is helpful to anyone who uses websockets in general... and is probably good template for anyone getting started in this area. I'd like to flesh this out into something that is more general purpose, and reusable, since the assumptions and questions I have will apply to many different situations.

Assumptions:

  1. The Timespan "Delay" is used to tell the server my client is OK and refreshes the subscription. Issuing this to the server too often is a waste of resources, and issuing it too late may cause a server-defined timeout.

  2. The receive chunksize is only how my client is talking to a network buffer. Delays in this do not affect the server. (unsure of this)

  3. The WebSocketStatus can be other values, such as closed, opening, etc., and this code isn't production ready since I don't handle those events.

Questions

  1. What must be done to make this code production ready?

  2. How do I determine the optimal chunk size for send and receive?

  3. How should I recover from an error in the code, the network, or a server error?

  4. How do I ensure that the data I'm getting is complete, and not just a fragment of data, cut off in mid-transition simply because the rest of the data is stuck in the buffer?

Code

  using System;
  using System.Collections.Generic;
  using System.Linq;
  using System.Net.WebSockets;
  using System.Text;
  using System.Threading;
  using System.Threading.Tasks;

  namespace BlockchainInfoWebSockets
  {
      class Client
      {
         private static object consoleLock = new object();
         private const int sendChunkSize = 256;
         private const int receiveChunkSize = 256;
         private const bool verbose = true;
         private static readonly TimeSpan delay = TimeSpan.FromMilliseconds(30000);

        static void Main(string[] args)
        {
            Thread.Sleep(1000);
            Connect("ws://ws.blockchain.info:8335/inv").Wait();
            Console.WriteLine("Press any key to exit...");
            Console.ReadKey();
        }

        public static async Task Connect(string uri)
        {
            ClientWebSocket webSocket = null;

            try
            {
                webSocket = new ClientWebSocket();
                await webSocket.ConnectAsync(new Uri(uri), CancellationToken.None);
                await Task.WhenAll(Receive(webSocket), Send(webSocket));
            }
            catch (Exception ex)
            {
                Console.WriteLine("Exception: {0}", ex);
            }
            finally
            {
                if (webSocket != null)
                    webSocket.Dispose();
                Console.WriteLine();

                lock (consoleLock)
                {
                    Console.ForegroundColor = ConsoleColor.Red;
                    Console.WriteLine("WebSocket closed.");
                    Console.ResetColor();
                }
            }
        }
       static UTF8Encoding encoder = new UTF8Encoding();

        private static async Task Send(ClientWebSocket webSocket)
        {

            //byte[] buffer = encoder.GetBytes("{\"op\":\"blocks_sub\"}"); //"{\"op\":\"unconfirmed_sub\"}");
            byte[] buffer = encoder.GetBytes("{\"op\":\"unconfirmed_sub\"}");
            await webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Text, true, CancellationToken.None);

          while (webSocket.State == WebSocketState.Open)
            {
                LogStatus(false, buffer, buffer.Length);
                await Task.Delay(delay);
            }
        }

        private static async Task Receive(ClientWebSocket webSocket)
        {
            byte[] buffer = new byte[receiveChunkSize];
            while (webSocket.State == WebSocketState.Open)
            {
                var result = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
                if (result.MessageType == WebSocketMessageType.Close)
                {
                    await webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None);
                }
                else
                {
                    LogStatus(true, buffer, result.Count);
                }
            }
        }

        private static void LogStatus(bool receiving, byte[] buffer, int length)
        {
            lock (consoleLock)
            {
                Console.ForegroundColor = receiving ? ConsoleColor.Green : ConsoleColor.Gray;
                //Console.WriteLine("{0} ", receiving ? "Received" : "Sent");

                if (verbose)
                    Console.WriteLine(encoder.GetString(buffer));

                Console.ResetColor();
            }
        }
    }
}
\$\endgroup\$
  • \$\begingroup\$ in Receive() we could add support for handling fragmentation - happens when the sent message is large ... \$\endgroup\$ – Sean Sep 13 '16 at 9:41
  • \$\begingroup\$ working on this myself right now .... specifically, we should check the return value from ReceiveAsync() class type: System.Net.WebSockets.WebSocketReceiveResult it has a flag to say whether this is the 'end' message OR is it a fragment: EndOfMessage [bool] \$\endgroup\$ – Sean Sep 13 '16 at 16:42
  • \$\begingroup\$ @Sean I'm not close to this code now, but please make edits for the greater good \$\endgroup\$ – goodguys_activate Sep 13 '16 at 17:20
  • \$\begingroup\$ In Receive method, shouldn't the byte[] buffer = new byte[receiveChunkSize]; be present inside the while loop ? \$\endgroup\$ – Pரதீப் Sep 25 at 5:09
5
\$\begingroup\$
  1. You may want to be using CancellationTokens via the overload for the async methods of the ClientWebSocket class to properly cleanup tasks on exit.
  2. 1024 is usually the value I like to use for web connectivity, the MTU (Maximum Transmission Unit)1 for Ethernet is 1500 bytes, and I like to round that down to a near exponent of 2, in this case 2^10.
  3. I'm not all sure about the ClientWebSocket class, but you will either get a WebException e.g. 404, I think in the StatusCode, or else it may be that WebSocketStatus you were talking about. Logging these exceptions may be the only thing you can do.
  4. Unless you have access to the source code of the server on the otherside and are able to make changes to it, unless the server implements some way to tell how much data it is sending, there's no way to be sure. An example would be the first 4 bytes of the transmission is the length of the data.
\$\endgroup\$
2
\$\begingroup\$

The wrapper sends and receives exactly one message before closing the connection. If you're just doing that, you can simply use HTTP. WebSockets is useful when there is a sequences of reads and/or writes with gaps of inactivity in between.

\$\endgroup\$
1
\$\begingroup\$

It's best practice to create thread synchronization objects using System.Threading.Interlocked.CompareEx. So something like:

private static object consoleLock
private static object GetLock() 
{ 
    System.Threading.Interlocked.CompareExchange(ref consoleLock, new object(), null);          
    return consoleLock; 
}

And I'd write an overload of LogStatus that accepts a string and a Color so you can clean up the socket close logging in your Connect method finally block.

\$\endgroup\$
  • 2
    \$\begingroup\$ Would you mind to provide a link or describe a value of using CompareExchange instead of relying on Framework initialization order? \$\endgroup\$ – pocheptsov Sep 4 '15 at 18:45

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.