.NET - 在CEF实现Server-Sent Events类型的资源控制器

前言

最近有群友向我请教如何在 CEF 中实现 Server-Sent Events（SSE） 类型的自定义资源控制器，目前主流的 AI 平台都采用了相似的技术对生成的内容进行流式返回，而该群友的需求是在 CEF 框架中对某个 URL 进行拦截并实时通过后端的 ASP.NET Core 流式返回 AICG 的内容。
之前我一直致力于 WinForm 与 Web 页面的结合，对 SSE 的知识了解甚少，通过 MDN 对《使用服务器发送事件》内容的学习，了解了其基本原理，其本质就是使用长连接保证了浏览器与服务器之间的流持续存在不断开，服务器根据后端的需求不定时向浏览器端发送事件（event）或者数据（message）等固定格式的 utf8 字符串，并以“\n\n”作为一条数据的终止信息。与 WebSokect 不同的是，这种通信方式是单向的，只能由服务器周期性向客户端发送数据，客户端使用 EventSource 实例来接收这些数据。有关于 Server-Sent Events 的相关内容请参考 MDN 的描述，在此不做过多介绍，本文的焦点将重点放在如何使用 CEF 的 CefRequestHandler 接口中的 GetResourceRequestHandler 方法通过返回自定义的 CefResourceRequestHandler 实例来进行请求拦截并实现一个简单的 SSE 长链接数据流。
根据该群友的需求需要基于 CefSharp 框架进行实现，因为平时本人比较熟悉的框架是 CefGlue 对 CefSharp 并不太熟悉，因此本示例如有纰漏还请各位读者之争。
实现

首先，用 Visual Studio 2022 创建一个“Windows 窗体应用程序”类型的项目，并使用 NuGet 包管理其安装 “CefShapr.WinForms” 包，版本根据实际需求进行选择。
然后，在默认的窗体 Form1.cs 的构造函数中添加 CefSharp 的浏览器控件，并把该浏览器控件实例的 RequestHandler 换成我们自己的请求控制器。

1. public partial class Form1 : Form
2. {
3.     private ChromiumWebBrowser browser;
4.     public Form1()
5.     {
6.         InitializeComponent();
8.         browser = new ChromiumWebBrowser();
9.         browser.Dock = DockStyle.Fill;
11.         this.Controls.Add(browser);
13.         browser.Load("https://test.local/sse");
15.         browser.RequestHandler = new MyRequestHandler();
16.     }
17. }

复制代码

在自定义的请求控制器 MyRequestHandler 里，我们只需要重载 GetResourceRequestHandler 方法，通过该重载方法提供自定义的 ResourceRequestHandler。

1. class MyRequestHandler : RequestHandler
2. {
3.     protected override IResourceRequestHandler GetResourceRequestHandler(IWebBrowser chromiumWebBrowser, IBrowser browser, IFrame frame, IRequest request, bool isNavigation, bool isDownload, string requestInitiator, ref bool disableDefaultHandling)
4.     {
5.       return new MyResourceRequestHandler();
6.     }
7. }

复制代码

在自定义的资源请求控制器 MyResourceRequestHandler 中，重载 GetResourceHandler 方法，在该方法中对指定 URL 进行拦截并将该请求映射到我们自己的 ResourceHandler 实例上，这样就能在该实例中实现我们的目标需求。

1. class MyResourceRequestHandler : ResourceRequestHandler
2. {
3.     protected override IResourceHandler GetResourceHandler(IWebBrowser chromiumWebBrowser, IBrowser browser, IFrame frame, IRequest request)
4.     {
5.         if(request.Url == "https://test.local/sse")
6.         {
7.             return new MyResourceHandler();
8.         }
10.         return base.GetResourceHandler(chromiumWebBrowser, browser, frame, request);
11.     }
12. }

复制代码

下面展示的代码将是本文的重点 MyResourceHandler 类，该类实现了 CefSharp 的 IResourceHandler 接口以此来实现一个完成的资源控制器，我们在该类的 Read 方法里用对客户端数据流进行阻塞实现长连接，并在需要的时候向该流中写入数据以实现 Server-Sent Events 的效果。
Sever-Sent Events 数据结构体
使用 ServerSentEvent 类来存储需要向客户端发送的 SSE 数据。

1. public sealed class ServerSentEvent
2. {
4.     public string Id { get; set; }
6.     public string Event { get; set; }
8.     public string Data { get; set; }
10.     public uint? Retry { get; set; }
12.     public override string ToString()
13.     {
14.         var sb = new StringBuilder();
15.         if (Retry.HasValue)
16.         {
17.             sb.Append($"retry: {Retry.Value}\n");
18.         }
20.         if (!string.IsNullOrWhiteSpace(Event))
21.         {
22.             sb.Append($"event: {Event}\n");
23.         }
24.         else
25.         {
26.             sb.Append("event: message\n");
27.         }
29.         if (!string.IsNullOrWhiteSpace(Id))
30.         {
31.             sb.Append($"id: {Id}\n");
32.         }
34.         var lines = Data?.Split(new[] { "\r\n", "\n" }, StringSplitOptions.RemoveEmptyEntries) ?? [""];
35.         foreach (var line in lines)
36.         {
37.             sb.Append($"data: {line}\n");
38.         }
39.         sb.Append("\n");
41.         return sb.ToString();
42.     }
43. }

复制代码

IResourceHandler 的实现
在这里如果继承 CefSharp 封装好的 ResourceHandler 类将不能很好的控制数据流，因此我们采用实现接口的方式使用更底层的代码实现对数据流的手动控制。MyResourceHandler 实现了 IResourceHandler 接口，用于自定义 CEF（Chromium Embedded Framework）资源处理，模拟 SSE（Server-Sent Events）流。

1. class MyResourceHandler : IResourceHandler
2. {
3.     // 用于异步等待新事件到来，配合 async/await 实现数据推送。
4.     TaskCompletionSource<bool> _taskCompletionSource = null;
5.     // 线程安全的事件队列，存储待发送的 ServerSentEvent。
6.     ConcurrentBag<ServerSentEvent> _events = new ConcurrentBag<ServerSentEvent>();
7.     // 标记连接是否活跃。
8.     bool _isConnected = true;
10.     // 连接断开，通知等待中的任务终止。
11.     public void Cancel()
12.     {
13.         _isConnected = false;
14.         _taskCompletionSource?.SetResult(false);
15.         _taskCompletionSource = null;
16.     }
18.     public void Dispose()
19.     {
20.     }
22.     public void GetResponseHeaders(IResponse response, out long responseLength, out string redirectUrl)
23.     {
24.         response.SetHeaderByName("X-Accel-Buffering", "no", true);
25.         response.SetHeaderByName("Content-Type", "text/event-stream,utf-8", true);
26.         response.SetHeaderByName("Cache-Control", "no-cache", true);
27.         response.SetHeaderByName("Access-Control-Allow-Origin", "*", true);
28.         response.MimeType = "text/event-stream";
29.         redirectUrl = string.Empty;
30.         responseLength = 0;
31.     }
33.     public bool Open(IRequest request, out bool handleRequest, ICallback callback)
34.     {
35.         _taskCompletionSource = new TaskCompletionSource<bool>();
37.         handleRequest = true;
39.         _isConnected = true;
41.         SimulateDataPushing();
43.         return true;
44.     }
46.     private async void SimulateDataPushing()
47.     {
48.         while (_isConnected)
49.         {
50.             await Task.Delay(1000); // Simulate a delay for pushing data
51.             var sse = new ServerSentEvent
52.             {
53.                 Id = Guid.NewGuid().ToString(),
54.                 Event = "message",
55.                 Data = $"Current time: {DateTime.Now:HH:mm:ss}",
56.          
57.             };
58.             _events.Add(sse);
59.             _taskCompletionSource?.TrySetResult(true);
60.         }
61.     }
63.     public bool Read(Stream dataOut, out int bytesRead, IResourceReadCallback callback)
64.     {
65.         HandleSSEAsync(dataOut, callback);
67.         bytesRead = 0;
68.         return true;
69.     }
71.     private async void HandleSSEAsync(Stream dataOut, IResourceReadCallback callback)
72.     {
74.         if (!_isConnected || _taskCompletionSource is null)
75.         {
76.             callback.Continue(0);
77.             return;
78.         }
80.         await _taskCompletionSource.Task;
82.         if (!_isConnected)
83.         {
84.             callback.Continue(0);
85.             return;
86.         }
88.         _taskCompletionSource = new TaskCompletionSource<bool>();
90.         if (_events.TryTake(out var msg))
91.         {
92.             var data = msg.ToString();
93.             var bytes = Encoding.UTF8.GetBytes(data);
94.             dataOut.Write(bytes, 0, bytes.Length);
95.             dataOut.Flush();
96.             callback.Continue(bytes.Length);
97.         }
98.     }
101.     #region NotImplementedException
103.     public bool ProcessRequest(IRequest request, ICallback callback)
104.     {
105.         throw new NotImplementedException();
106.     }
109.     public bool ReadResponse(Stream dataOut, out int bytesRead, ICallback callback)
110.     {
111.         throw new NotImplementedException();
112.     }
114.     public bool Skip(long bytesToSkip, out long bytesSkipped, IResourceSkipCallback callback)
115.     {
116.         bytesSkipped = bytesToSkip;
117.         return true;
118.     }
119.     #endregion
120. }

复制代码

上面代码主要进行了以下操作：

• GetResponseHeaders 方法为浏览器请求设置了 HTTP 响应头，声明 SSE 流（text/event-stream），禁止缓存，并允许跨域访问。
  
• Open 方法将在客户端连接的时候被触发，在示例中，按照 CEF 的官方文档，让参数 handleRequest 为 true，并返回 true 表示立即处理该请求。其中的异步方法 SimulateDataPushing 实现的是数据推送的模拟，这个模拟每秒模拟生成一个新的服务器事件，加入队列，并唤醒等待的读取操作。
  
• Read 方法将在数据流被读取时被触发，这也是实现长连接的关键。这里按照 CEF 的官方文档，让参数 bytesRead 为 0，并返回 true 表示异步处理 Read 请求，这里使用异步方法 HandleSSEAsync 来处理队列 _events 里的数据，当有新事件到来后，将事件内容写入输出流，通知 CEF 继续读取，读取完成后新建 TaskCompletionSource 继续阻塞改方法。
  

以上，就完成了 SSE 类型资源控制器的实现。启动项目，直接访问该拦截地址 https://test.local/sse，实现了长连接并按预期收到了周期性的服务器发送事件。

结论

对以上代码的扩展，接入了 AICG 后浏览器作为 SSE 客户端正常收到了我们自定义的 SSE 资源控制器发送过来的数据。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！