source: binary-improvements2/WebServer/src/SSE/AbsEvent.cs@ 438

using System;
using System.Collections.Generic;
using System.IO;
using System.Net.Sockets;
using System.Text;
using Webserver.UrlHandlers;
using HttpListenerResponse = SpaceWizards.HttpListener.HttpListenerResponse;

namespace Webserver.SSE {
        public abstract class AbsEvent {
                private const int encodingBufferSize = 1024 * 1024;
                private const int keepAliveIntervalSeconds = 10;
                private static readonly byte[] keepAliveData = Encoding.UTF8.GetBytes (": KeepAlive\n\n");

                private readonly SseHandler parent;
                public readonly string Name;

                private readonly byte[] encodingBuffer;
                private readonly StringBuilder stringBuilder = new StringBuilder ();
                private DateTime lastMessageSent;

                private readonly List<HttpListenerResponse> openStreams = new List<HttpListenerResponse> ();

                private readonly BlockingQueue<(string _eventName, string _data)> sendQueue =
                        new BlockingQueue<(string _eventName, string _data)> ();

                private int currentlyOpen;
                private int totalOpened;
                private int totalClosed;

                protected AbsEvent (SseHandler _parent, bool _reuseEncodingBuffer = true, string _name = null) {
                        Name = _name ?? GetType ().Name;
                        parent = _parent;
                        if (_reuseEncodingBuffer) {
                                encodingBuffer = new byte[encodingBufferSize];
                        }
                }

                public void AddListener (HttpListenerResponse _resp) {
                        totalOpened++;
                        currentlyOpen++;

                        openStreams.Add (_resp);
                }

                protected void SendData (string _eventName, string _data) {
                        sendQueue.Enqueue ((_eventName, _data));
                        parent.SignalSendQueue ();
                }

                public void ProcessSendQueue () {
                        bool dataSent = false;
                        
                        while (sendQueue.HasData ()) {
                                (string eventName, string data) = sendQueue.Dequeue ();
                                
                                stringBuilder.Append ("event: ");
                                stringBuilder.AppendLine (eventName);
                                stringBuilder.Append ("data: ");
                                stringBuilder.AppendLine (data);
                                
                                stringBuilder.AppendLine ("");

                                string output = stringBuilder.ToString ();
                                stringBuilder.Clear ();

                                byte[] buf;
                                int bytesToSend;
                                if (encodingBuffer != null) {
                                        buf = encodingBuffer;
                                        try {
                                                bytesToSend = Encoding.UTF8.GetBytes (output, 0, output.Length, buf, 0);
                                        } catch (ArgumentException e) {
                                                logError ("Exception while encoding data for output, most likely exceeding buffer size:", false);
                                                Log.Exception (e);
                                                return;
                                        }
                                } else {
                                        buf = Encoding.UTF8.GetBytes (output);
                                        bytesToSend = buf.Length;
                                }

                                dataSent = true;

                                sendBufToListeners (buf, bytesToSend);
                        }

                        DateTime now = DateTime.Now;
                        if (dataSent) {
                                lastMessageSent = now;
                        } else if ((now - lastMessageSent).TotalSeconds >= keepAliveIntervalSeconds) {
                                sendBufToListeners (keepAliveData, keepAliveData.Length);
                                lastMessageSent = now;
                        }
                }

                private void sendBufToListeners (byte[] _bytes, int _bytesToSend) {
                        for (int i = openStreams.Count - 1; i >= 0; i--) {
                                HttpListenerResponse resp = openStreams [i];
                                try {
                                        if (resp.OutputStream.CanWrite) {
                                                resp.OutputStream.Write (_bytes, 0, _bytesToSend);
                                                resp.OutputStream.Flush ();
                                        } else {
                                                currentlyOpen--;
                                                totalClosed++;

                                                logError ("Can not write to endpoint, closing", true);
                                                openStreams.RemoveAt (i);
                                                resp.Close ();
                                        }
                                } catch (IOException e) {
                                        currentlyOpen--;
                                        totalClosed++;

                                        openStreams.RemoveAt (i);

                                        if (e.InnerException is SocketException se) {
                                                if (se.SocketErrorCode != SocketError.ConnectionAborted && se.SocketErrorCode != SocketError.Shutdown) {
                                                        logError ($"SocketError ({se.SocketErrorCode.ToStringCached ()}) while trying to write", true);
                                                }
                                        } else {
                                                logError ("IOException while trying to write:", true);
                                                Log.Exception (e);
                                        }
                                } catch (Exception e) {
                                        currentlyOpen--;
                                        totalClosed++;

                                        openStreams.RemoveAt (i);
                                        logError ("Exception while trying to write:", true);
                                        Log.Exception (e);
                                        resp.Close ();
                                }
                        }
                }

                protected void logError (string _message, bool _printConnections) {
                        Log.Error (_printConnections
                                ? $"[Web] [SSE] '{Name}': {_message} (Left open: {currentlyOpen}, total opened: {totalOpened}, closed: {totalClosed})"
                                : $"[Web] [SSE] '{Name}': {_message}");
                }

                public virtual int DefaultPermissionLevel () => 0;
        }
}