WebSockets and Server-sent events

 0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-------+-+-------------+-------------------------------+
|F|R|R|R| opcode|M| Payload len |    Extended payload length    |
|I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
|N|V|V|V|       |S|             |   (if payload len==126/127)   |
| |1|2|3|       |K|             |                               |
+-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
|     Extended payload length continued, if payload len == 127  |
+ - - - - - - - - - - - - - - - +-------------------------------+
|                               |Masking-key, if MASK set to 1  |
+-------------------------------+-------------------------------+
| Masking-key (continued)       |          Payload Data         |
+-------------------------------- - - - - - - - - - - - - - - - +
:                     Payload Data continued ...                :
+ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
|                     Payload Data continued ...                |
+---------------------------------------------------------------+

FIN:  1 bit

   Indicates that this is the final fragment in a message.  The first
   fragment MAY also be the final fragment.

Opcode:  4 bits

   Defines the interpretation of the "Payload data".  If an unknown
   opcode is received, the receiving endpoint MUST _Fail the
   WebSocket Connection_.  The following values are defined.

   *  %x0 denotes a continuation frame

   *  %x1 denotes a text frame

   *  %x2 denotes a binary frame

   *  %x3-7 are reserved for further non-control frames

   *  %x8 denotes a connection close

   *  %x9 denotes a ping

   *  %xA denotes a pong

   *  %xB-F are reserved for further control frames

var webSocketOptions = new WebSocketOptions
{
    KeepAliveInterval = TimeSpan.FromMinutes(2)
};

webSocketOptions.AllowedOrigins.Add("https://client.com");
webSocketOptions.AllowedOrigins.Add("https://www.client.com");

// Add the WebSockets middleware in `Program.cs`:
app.UseWebSockets(webSocketOptions);

app.Use(async (context, next) =>
{
    // [Route("/ws")] // HTTP/2 WebSockets use CONNECT requests rather than GET.
    if (context.Request.Path == "/ws")
    {
        // Accept WebSocket requests
        if (context.WebSockets.IsWebSocketRequest)
        {
            using var webSocket = await context.WebSockets.AcceptWebSocketAsync();
            await EchoAsync(webSocket);
        }
        else
        {
            context.Response.StatusCode = StatusCodes.Status400BadRequest;
        }
    }
    else
    {
        await next(context);
    }

});

app.Run();

// Send and receive messages
static async Task EchoAsync(WebSocket webSocket)
{
    var buffer = new byte[1024 * 4];
    var receiveResult = await webSocket.ReceiveAsync(
        new ArraySegment<byte>(buffer), CancellationToken.None);

    while (!receiveResult.CloseStatus.HasValue)
    {
        await webSocket.SendAsync(
            new ArraySegment<byte>(buffer, 0, receiveResult.Count),
            receiveResult.MessageType,
            receiveResult.EndOfMessage,
            CancellationToken.None);

        receiveResult = await webSocket.ReceiveAsync(
            new ArraySegment<byte>(buffer), CancellationToken.None);
    }

    await webSocket.CloseAsync(
        receiveResult.CloseStatus.Value,
        receiveResult.CloseStatusDescription,
        CancellationToken.None);
}

string[] messages = [
    "我们的生命不是消逝于那些重大的事件中，而是流逝在那些日常琐碎的小事里。",
    "生活是由无数微不足道的细节构建起来的，而记忆正是这些细节的忠实记录者，它们在某个不经意的瞬间被唤醒，带我们穿越回往昔。",
    "人们在追求他们以为是幸福的东西时，常常错过真正的幸福。",
    "在失去之后，我们才开始寻找那些曾经拥有但未被珍惜的东西，而记忆，则成了我们找回那些失落时光的唯一线索。"
    ];

Uri uri = new("ws://localhost:5000/ws");
using ClientWebSocket ws = new();
var cts = new CancellationTokenSource();
await ws.ConnectAsync(uri, cts.Token);

foreach (var message in messages)
{
    var bytes = Encoding.UTF8.GetBytes(message);
    await ws.SendAsync(bytes, WebSocketMessageType.Text, true, cts.Token);
}

ThreadPool.QueueUserWorkItem(async _ =>
{
    while (!cts.Token.IsCancellationRequested)
    {
        var (echoMessage, _, _, _, _) = await ReadMessageAsync(ws, cts.Token);
        Console.WriteLine(Encoding.UTF8.GetString(echoMessage.ToArray()));
    }
});

await Task.Delay(1000);

await ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "Client closed", cts.Token);

// Read a complete message from a WebSocket.
static async Task<(IList<byte>, WebSocketMessageType, bool, WebSocketCloseStatus?, string?)> ReadMessageAsync(WebSocket webSocket, CancellationToken token = default)
{
    var message = new List<byte>(1024 * 2);
    var buffer = new byte[8 * 4];
    var receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), token).ConfigureAwait(false);
    while (true)
    {
        message.AddRange(new ArraySegment<byte>(buffer, 0, receiveResult.Count));
        if (receiveResult.CloseStatus.HasValue || receiveResult.EndOfMessage)
        {
            break;
        }
        receiveResult = await webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), token).ConfigureAwait(false);
    }

    return (message.AsReadOnly(), receiveResult.MessageType, receiveResult.EndOfMessage, receiveResult.CloseStatus, receiveResult.CloseStatusDescription);
}

const excerpts = [
  'Grown-ups never understand anything by themselves, and it is tiresome for children to be always and forever explaining things to them.',
  'And now here is my secret, a very simple secret: It is only with the heart that one can see rightly; what is essential is invisible to the eye.',
  "People have forgotten this truth,' the fox said. 'But you mustn’t forget it. You become responsible forever for what you’ve tamed. You’re responsible for your rose.",
  'All grown-ups were once children... but only few of them remember it.',
  'It is the time you have wasted for your rose that makes your rose so important.',
  'One sees clearly only with the heart. Anything essential is invisible to the eyes.',
  'You - you alone will have the stars as no one else has them... In one of the stars I shall be living. In one of them I shall be laughing. And so it will be as if all the stars were laughing, when you look at the sky at night... You - only you - will have stars that can laugh.',
  'You become responsible, forever, for what you have tamed.'
]

// Creating a WebSocket object
const ws = new WebSocket('ws://localhost:5000/ws')
// Listen for possible errors
ws.addEventListener('error', (event) => {
  console.log('WebSocket error: ', event)
})

// Sending data to the server
ws.onopen = () => {
  for (const excerpt of excerpts) {
    ws.send(excerpt)
  }
}

// Receiving messages from the server
ws.onmessage = (e) => {
  console.log(e.data)
}

app.UseCors(policy => policy.AllowAnyOrigin()); // builder.Services.AddCors();

var excerpts = new string[]
{
  "Notre vie ne se gaspille pas dans les grands événements, mais s'écoule dans les petites choses de tous les jours.",
  "La vie est faite de millions de détails insignifiants, et la mémoire est le fidèle enregistreur de ces détails, qui se réveillent à un moment inattendu et nous transportent dans le passé.",
  "En poursuivant ce qu'ils croient être le bonheur, les gens passent souvent à côté du vrai bonheur.",
  "Ce n'est qu'après avoir perdu quelque chose que nous commençons à chercher ce que nous avions et que nous n'avons pas chéri, et la mémoire devient alors le seul fil conducteur pour retrouver ces moments perdus."
};

app.Use(async (context, next) =>
{
    if (context.Request.Path == "/sse")
    {
        if (context.Request.Headers.Accept.Any(x => x != null && x.Contains("text/event-stream")))
        {
            context.Response.Headers.ContentType = "text/event-stream";
            context.Response.Headers.CacheControl = "no-cache";

            await context.Response.Body.WriteAsync(System.Text.Encoding.UTF8.GetBytes($"event: ping\ndata: pong!\n\n"));
            await context.Response.Body.FlushAsync();

            foreach (var excerpt in excerpts)
            {
                await context.Response.Body.WriteAsync(System.Text.Encoding.UTF8.GetBytes($"data: {excerpt}\n\n"));
                await context.Response.Body.FlushAsync();
            }

            // the stream terminated by a data: [DONE]
            await context.Response.Body.WriteAsync(System.Text.Encoding.UTF8.GetBytes("data: [DONE]\n\n"));
            await context.Response.Body.FlushAsync();
        }
        else
        {
            context.Response.StatusCode = StatusCodes.Status415UnsupportedMediaType;
        }
    }
    else
    {
        await next(context);
    }
});

app.Run();

$ curl -i localhost:5000/sse -H 'accept: text/event-stream'
HTTP/1.1 200 OK
Content-Type: text/event-stream
Date: Thu, 16 May 2024 11:14:26 GMT
Server: Kestrel
Cache-Control: no-cache
Transfer-Encoding: chunked

event: ping
data: pong!

data: Notre vie ne se gaspille pas dans les grands événements, mais s'écoule dans les petites choses de tous les jours.

data: La vie est faite de millions de détails insignifiants, et la mémoire est le fidèle enregistreur de ces détails, qui se réveillent à un moment inattendu et nous transportent dans le passé.

data: En poursuivant ce qu'ils croient être le bonheur, les gens passent souvent à côté du vrai bonheur.

data: Ce n'est qu'après avoir perdu quelque chose que nous commençons à chercher ce que nous avions et que nous n'avons pas chéri, et la mémoire devient alors le seul fil conducteur pour retrouver ces moments perdus.

data: [DONE]

// Creating an EventSource instance
const evtSource = new EventSource('//localhost:5000/sse', {
  // withCredentials: true
})
// Listening for message events
evtSource.onmessage = (event) => {
  console.log(`message: ${event.data}`)
  // the stream terminated by a data: [DONE]
  if (event.data === '[DONE]') {
    // Closing event streams
    evtSource.close()
  }
}
// Listening for custom events
evtSource.addEventListener('ping', (event) => {
  console.log(`ping: ${event.data}`)
})
// Error handling
evtSource.onerror = (err) => {
  console.error('EventSource failed:', err)
}