C#实现SSE(Server-Sent Events)服务端和客户端的示例代码

using system;
using system.collections.generic;
using system.net;
using system.text;
using system.threading;
 
namespace sseserver
{
    class program
    {
        static void main(string[] args)
        {
            httplistener listener = new httplistener();
            listener.prefixes.add("http://localhost:8080/");
            listener.start();
 
            console.writeline("server is running...");
 
            while (true)
            {
                httplistenercontext context = listener.getcontext();
                threadpool.queueuserworkitem((_) =>
                {
                    sendevent(context);
                });
            }
        }
 
        static void sendevent(httplistenercontext context)
        {
            try
            {
                context.response.contenttype = "text/event-stream";
                context.response.statuscode = 200;
                context.response.headers.add("cache-control", "no-cache");
                context.response.headers.add("connection", "keep-alive");
 
                while (true)
                {
                    string message = $"data: {datetime.now.tostring("hh:mm:ss")}\n\n";
                    byte[] bytes = encoding.utf8.getbytes(message);
                    context.response.outputstream.write(bytes, 0, bytes.length);
                    context.response.outputstream.flush();
                    thread.sleep(1000); // 每秒发送一次消息
                }
            }
            catch (exception ex)
            {
                console.writeline("error: " + ex.message);
            }
            finally
            {
                context.response.close();
            }
        }
    }
}

using system;
using system.io;
using system.net;
using system.text;
namespace sseclient
{
    class program
    {
        static async task main(string[] args)
        {
            string url = "http://localhost:8080/sse_endpoint"; // sse endpoint url
 
            using (httpclient client = new httpclient())
            {
                var request = new httprequestmessage(httpmethod.get, url);
                request.headers.add("accept", "text/event-stream");
 
                using (httpresponsemessage response = await client.sendasync(request, httpcompletionoption.responseheadersread))
                {
                    response.ensuresuccessstatuscode();
 
                    using (var responsestream = await response.content.readasstreamasync())
                    using (var reader = new system.io.streamreader(responsestream))
                    {
                        while (!reader.endofstream)
                        {
                            string line = await reader.readlineasync();
 
                            if (!string.isnullorwhitespace(line))
                            {
                                console.writeline(line);
                                // handle the received sse event here
                            }
                        }
                    }
                }
            }
        }
    }
}

// 建立serversentevents（服务器向浏览器推送信息，简称sse）
    $(function () {
        if (typeof (eventsource) !== "undefined") {
            // 展示服务器推送内容的dom
            var container = document.getelementbyid("ssecontainer");
            // 建立sse通道
            var es = new eventsource("/api/serversentevents/buildingsse");
            // 监听sse通道open事件
            es.onopen = function (event) {
                container.innerhtml += "open<br/>";
            }
            // 监听sse接收到的服务端消息
            es.onmessage = function (event) {
                container.innerhtml += event.data + "<br/>";
            };
            // 监听sse通道error事件
            es.onerror = function (event) {
                container.innerhtml += "error<br/>";
                es.close();
            }
        }
        else {
            console.log("浏览器不支持serversentevents接口!");
        }
    })

using newtonsoft.json;
using system;
using system.collections.concurrent;
using system.io;
using system.net;
using system.net.http;
using system.net.http.headers;
using system.web.http;

namespace cabinetms.controllers.webapi
{

    /// <summary>
    /// 自定义的sse消息对象实体
    /// </summary>
    public class ssemessageobject
    {
        public string msgid { get; set; }
        public string msgdata { get; set; }
    }



    /// <summary>
    /// 以请求方式建立sse通道
    /// </summary>
    [routeprefix("api/serversentevents")]
    public class serversenteventcontroller : apicontroller
    {

        // 接收浏览器请求，建立serversentevents通道
        [httpget, route("buildingsse")]
        public system.net.http.httpresponsemessage buildingsse(httprequestmessage request)
        {
            system.net.http.httpresponsemessage response = request.createresponse();
            response.content = new system.net.http.pushstreamcontent((action<stream, httpcontent, transportcontext>)writetostream, new mediatypeheadervalue("text/event-stream"));
            return response;
        }

        private static readonly concurrentdictionary<streamwriter, streamwriter> _streammessage = new concurrentdictionary<streamwriter, streamwriter>();
        public void writetostream(stream outputstream, httpcontent content, transportcontext context)
        {
            streamwriter streamwriter = new streamwriter(outputstream);
            _streammessage.tryadd(streamwriter, streamwriter);
        }

        // 建立sse通道后，其他controller或程序调用此方法，可以向浏览器端主动推送消息
        public static void sendssemsg(ssemessageobject ssemsg)
        {
            messagecallback(ssemsg);
        }

        // 设置向浏览器推送的消息内容
        private static void messagecallback(ssemessageobject ssemsg)
        {
            foreach (var subscriber in _streammessage.toarray())
            {
                try
                {
                    subscriber.value.writeline(string.format("id: {0}\n", ssemsg.msgid));
                    subscriber.value.writeline(string.format("data: {0}\n\n", ssemsg.msgdata));
                    subscriber.value.flush();
                }
                catch
                {
                    streamwriter streamwriter;
                    _streammessage.tryremove(subscriber.value, out streamwriter);
                }
            }
        }
    }
}

// 服务端向网页端推送告警信息
var ssemsg = new ssemessageobject();
ssemsg.msgid = "1101";
ssemsg.msgdata = "自定义告警消息";
serversenteventcontroller.sendssemsg(ssemsg);