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:
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.
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)
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
What must be done to make this code production ready?
How do I determine the optimal chunk size for send and receive?
How should I recover from an error in the code, the network, or a server error?
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();
}
}
}
}
Receive
method, shouldn't thebyte[] buffer = new byte[receiveChunkSize];
be present inside the while loop ? \$\endgroup\$