NET CORE 2.1和EventSource示例

缺乏、安全感 2021-06-26 16:06 394阅读 0赞

## 项目结构 ##

![在这里插入图片描述][20190725172142494.png]

## HomeController.cs ##

using System;
    using System.Collections.Generic;
    using System.IO;
    using System.Linq;
    using System.Text;
    using System.Threading;
    using System.Threading.Tasks;
    using EventSourceDemo.Services;
    using Microsoft.AspNetCore.Mvc;
    using Microsoft.Extensions.Logging;
    
    namespace EventSourceDemo.Controllers
    { 
        public class HomeController : Controller
        { 
            public HomeController(EventsService eventsService,
                ILogger<HomeController> logger)
            { 
                //依赖注入事件服务和日志记录器
                // dependency inject the event service and logger
                this.eventsService = eventsService;
                this.logger = logger;
    
                //理想情况下，这应该通过某种计时器服务来处理这种单一计时器，这种计时器服务可以通过DI推送到这里
                // create a singleton timer in a slightly hacky fashion
                // ideally this should be handled by some sort of timer service
                // that gets pushed into here with DI
                lock (heartbeatTimerLock)
                { 
                    if (heartbeatTimer == null)
                    { 
                        heartbeatTimer = new Timer(new TimerCallback(HeartbeatTimerTick), null, 1000, 1000);
                    }
                }
            }
    
            private EventsService eventsService;
            private ILogger<HomeController> logger;
    
            private static readonly object heartbeatTimerLock = new object();
            private volatile static Timer heartbeatTimer = null;
    
            [HttpGet]
            public IActionResult Index()
            { 
                return View();
            }
    
            private void HeartbeatTimerTick(object state)
            { 
                //每次计时器触发时，它都会发出一个心跳事件，其事件源更新包含当前时间作为字符串
                // every time the timer fires, it will raise a heartbeat event
                // with an event source update that contains the current time
                // as a string
                string currentTimeString = DateTimeOffset.Now.ToString("o");
    
                eventsService.Notify("heartbeat", new EventSourceUpdate()
                { 
                    Comment = $"heartbeat {currentTimeString}",
                    Event = "Message",
                    DataObject = new { Message = currentTimeString}
                });
            }
    
            [HttpGet]
            public async Task<string> EventSource()
            { 
                //根据规范使用事件流内容类型
                // use the event stream content type, as per specification
                HttpContext.Response.ContentType = "text/event-stream";
    
                //从标题中获取最后一个事件ID
                // get the last event id out of the header
                string lastEventIdString = HttpContext.Request.Headers["Last-Event-ID"].FirstOrDefault();
                int temp;
                int? lastEventId = null;
    
                if (lastEventIdString != null && int.TryParse(lastEventIdString, out temp))
                { 
                    lastEventId = temp;
                }
    
                string remoteIp = HttpContext.Connection.RemoteIpAddress.ToString();
    
    
                // open the current request stream for writing.
                // Use UTF-8 encoding, and do not close the stream when disposing.
                using (var clientStream =
                    new StreamWriter(HttpContext.Response.Body, Encoding.UTF8, 1024, true) { AutoFlush = true})
                { 
                    // subscribe to the heartbeat event. Elsewhere, a timer will push updates to this event periodically.
                    using (EventSubscription<EventSourceUpdate> subscription =
                        eventsService.SubscribeTo<EventSourceUpdate>("heartbeat"))
                    { 
                        try
                        { 
                            logger.LogInformation($"Opened event source stream to address: {remoteIp}");
    
                            await clientStream.WriteLineAsync($":connected {DateTimeOffset.Now.ToString("o")}");
    
                            // If a last event id is given, pump out any intermediate events here.
                            if (lastEventId != null)
                            { 
                                // We're not doing anything that stores events in this heartbeat demo,
                                // so do nothing here.
                            }
    
                            // start pumping out events as they are pushed to the subscription queue
                            while (true)
                            { 
                                // asynchronously wait for an event to fire before continuing this loop.
                                // this is implemented with a semaphore slim using the async wait, so it
                                // should play nice with the async framework.
                                EventSourceUpdate update = await subscription.WaitForData();
    
                                // push the update down the request stream to the client
                                if (update != null)
                                { 
                                    string updateString = update.ToString();
                                    await clientStream.WriteAsync(updateString);
                                }
                            }
                        }
                        catch (Exception e)
                        { 
                            // catch client closing the connection
                            logger.LogInformation($"Closed event source stream from {remoteIp}. Message: {e.Message}");
                        }
                    }
                }
    
                return ":closed";
            }
    
            public IActionResult Contact()
            { 
                ViewData["Message"] = "Your contact page.";
    
                return View();
            }
    
            public IActionResult Error()
            { 
                return View();
            }
        }
    }

## Services ##

EventService.cs

using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;
    
    namespace EventSourceDemo.Services
    { 
        public class EventsService
        { 
            private readonly object collectionsLock = new object();
    
            //此字典将事件名称链接到该事件的订阅者列表
            // This dictionary links event names to a list of subscribers to that event
            private Dictionary<string, List<EventSubscription>> eventSubscriptionList =
                new Dictionary<string, List<EventSubscription>>();
    
            //此字典将订阅者链接到他们订阅的列表，以便在取消订阅时快速删除
            // this dictionary links subscribers to the lists that they are subscribed to,
            // for quick removal during unsubscribe
            private Dictionary<EventSubscription, List<List<EventSubscription>>> subscriptionMembershipLists =
                new Dictionary<EventSubscription, List<List<EventSubscription>>>();
    
            public void Notify(string eventName, object data)
            { 
                lock (collectionsLock)
                { 
                    List<EventSubscription> subscriptions = null;
                    if (eventSubscriptionList.TryGetValue(eventName, out subscriptions))
                    { 
                        //通知所有订阅此事件已发生
                        // notify all subscriptions that this event has occurred
                        NotifyList(subscriptions, data);
                    }
                }
            }
    
            public void NotifyWhere(Func<string, bool> predicate, object data)
            { 
                lock (collectionsLock)
                { 
                    List<EventSubscription> subscriptions = eventSubscriptionList.Where(kvp => predicate(kvp.Key))
                        .SelectMany(kvp => kvp.Value)
                        .Distinct()
                        .ToList();
    
                    NotifyList(subscriptions, data);
                }
            }
    
            private void NotifyList(List<EventSubscription> subscriptions, object data)
            { 
                foreach (EventSubscription thisSubscription in subscriptions)
                { 
                    thisSubscription.Notify(data);
                }
            }
    
            public EventSubscription<T> SubscribeTo<T>(params string[] eventNames) where T : class
            { 
                if (eventNames == null) throw new ArgumentNullException(nameof(eventNames));
                EventSubscription<T> subscription = new EventSubscription<T>(this);
    
                lock (collectionsLock)
                { 
                    List<List<EventSubscription>> eventMemberships = new List<List<EventSubscription>>();
                    subscriptionMembershipLists.Add(subscription, eventMemberships);
    
                    foreach (string thisEventName in eventNames)
                    { 
                        List<EventSubscription> subscriptions = null;
                        if (!eventSubscriptionList.TryGetValue(thisEventName, out subscriptions))
                        { 
                            //如果该事件名称不存在，请添加该列表
                            // add the list against this event name if it doesn't exist
                            subscriptions = new List<EventSubscription>();
                            eventSubscriptionList.Add(thisEventName, subscriptions);
                        }
    
                        //将订阅添加到此订阅列表
                        // add the subscription to this subscriptions list
                        subscriptions.Add(subscription);
                        //将此订阅列表添加到此属于的事件成员资格列表中。
                        // add this subscriptions list to the list of event memberships that this is part of.
                        eventMemberships.Add(subscriptions);
                    }
                }
    
                return subscription;
            }
    
            public void Unsubscribe(EventSubscription subscription)
            { 
                lock (collectionsLock)
                { 
                    List<List<EventSubscription>> subscriptionsList;
                    if (subscriptionMembershipLists.TryGetValue(subscription, out subscriptionsList))
                    { 
                        foreach (List<EventSubscription> subscriptions in subscriptionsList)
                        { 
                            lock (subscriptions)
                            { 
                                subscriptions.Remove(subscription);
                            }
                        }
    
                        subscriptionMembershipLists.Remove(subscription);
                    }
                }
            }
        }
    
        public class DataReceivedEventArgs<T> : EventArgs
        { 
            public DataReceivedEventArgs(T data)
            { 
            }
    
            public T Data {  get; }
        }
    
        public abstract class EventSubscription : IDisposable
        { 
            public abstract void Dispose();
    
            public abstract void Notify(object data);
        }
    
        public class EventSubscription<T> : EventSubscription where T : class
        { 
            public event EventHandler<DataReceivedEventArgs<T>> DataReceived;
            private EventsService service;
            private readonly object deliveryQueueLock = new object();
            private Queue<T> deliveryQueue = new Queue<T>();
            private SemaphoreSlim signal = new SemaphoreSlim(0, int.MaxValue);
    
            public EventSubscription(EventsService service)
            { 
                this.service = service;
            }
    
            public override void Notify(object data)
            { 
                T genericData = data as T;
    
                if (genericData != null)
                { 
                    Notify(genericData);
                }
            }
    
            public void Notify(T data)
            { 
                //将数据添加到队列以供使用
                // add data to the queue for consumption
                lock (deliveryQueueLock)
                { 
                    deliveryQueue.Enqueue(data);
                }
    
                signal.Release();
    
                //触发事件当发生事件与数据
                // fire event occured event with data
                DataReceived?.Invoke(this, new DataReceivedEventArgs<T>(data));
            }
    
            public async Task<T> WaitForData()
            { 
                await signal.WaitAsync();
    
                lock (deliveryQueueLock)
                { 
                    return deliveryQueue.Dequeue();
                }
            }
    
            public override void Dispose()
            { 
                service.Unsubscribe(this);
                signal.Dispose();
            }
        }
    }

EventSourceUpdate.cs

using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading.Tasks;
    using Newtonsoft.Json;
    
    namespace EventSourceDemo.Services
    { 
        public class EventSourceUpdate
        { 
            public int? Id {  get; set; }
            public string Event {  get; set; }
    
            private object dataObject;
    
            public object DataObject
            { 
                get {  return this.dataObject; }
                set
                { 
                    this.dataObject = value;
                    this.Data = JsonConvert.SerializeObject(value)
                        .Split(new string[] { Environment.NewLine}, StringSplitOptions.RemoveEmptyEntries)
                        .ToList();
                }
            }
    
            public List<string> Data {  get; private set; }
            public string Comment {  get; set; }
    
            public override string ToString()
            { 
                string idStart = "id:";
                string eventStart = "event:";
                string dataStart = "data:";
                string commentStart = ":";
    
                StringBuilder sb = new StringBuilder();
    
                if (Comment != null)
                { 
                    sb.Append(commentStart);
                    sb.AppendLine(Comment);
                }
    
                if (Event != null)
                { 
                    if (Id != null)
                    { 
                        sb.Append(idStart);
                        sb.AppendLine(Id.Value.ToString());
                    }
    
                    sb.Append(eventStart);
                    sb.AppendLine(Event);
    
                    if (Data != null)
                    { 
                        foreach (string thisDatum in Data)
                        { 
                            sb.Append(dataStart);
                            sb.AppendLine(thisDatum);
                        }
                    }
    
                    sb.AppendLine();
                }
    
                return sb.ToString();
            }
    
            public static EventSourceUpdate FromObject(string eventName, object value, int? id = null)
            { 
                return new EventSourceUpdate()
                { 
                    Id = id,
                    Event = eventName,
                    DataObject = value
                };
            }
        }
    }

## Index.cshtml ##

@{
        ViewData["Title"] = "Home Page";
    }
    <br />
    <div id="display">
    
    </div>
    @section Scripts {
        <script> // setup the event source function connectEventSource() { console.log("连接到事件源event source ..."); var eventSourceURL = "@Html.Raw(Url.Action("EventSource", "Home"))"; if (eventSourceURL !== "") { if (!!window.EventSource) { var source = new EventSource(eventSourceURL); console.log("已配置事件源."); // event source successfully connected source.addEventListener('Message', function (e) { dataObject = JSON.parse(e.data); var message = dataObject.Message; console.log("收到消息: " + message); writeLine(message); }, false); source.addEventListener('open', function (e) { // Connection was opened. console.log("事件源连接已打开."); writeLine("事件源已打开"); }, false); source.addEventListener('error', function (e) { if (e.readyState == EventSource.CLOSED) { // Connection was closed. console.log("事件源连接已关闭."); writeLine("事件源已关闭"); } }, false); } else { console.error("您的浏览器不支持事件源."); writeLine("浏览器不支持事件源SSE."); } } else { console.error("事件源URL无效"); } } function writeLine(message) { $("#display").append(message + "<br />"); } $(document).ready(function () { console.log("Document准备好了"); connectEventSource(); }); </script>
    
    }

## \_Layout.cshtml ##

<!DOCTYPE html>
    <html>
    <head>
        <meta charset="utf-8"/>
        <meta name="viewport" content="width=device-width, initial-scale=1.0"/>
        <title>@ViewData["Title"] - EventSource示例</title>
    
        <environment names="Development">
            <link rel="stylesheet" href="~/lib/bootstrap/dist/css/bootstrap.css"/>
            <link rel="stylesheet" href="~/css/site.css"/>
        </environment>
        <environment names="Staging,Production">
            <link rel="stylesheet" href="https://ajax.aspnetcdn.com/ajax/bootstrap/3.3.6/css/bootstrap.min.css" asp-fallback-href="~/lib/bootstrap/dist/css/bootstrap.min.css" asp-fallback-test-class="sr-only" asp-fallback-test-property="position" asp-fallback-test-value="absolute"/>
            <link rel="stylesheet" href="~/css/site.min.css" asp-append-version="true"/>
        </environment>
    </head>
    <body>
    <div class="navbar navbar-inverse navbar-fixed-top">
        <div class="container">
            <div class="navbar-header">
                <button type="button" class="navbar-toggle" data-toggle="collapse" data-target=".navbar-collapse">
                    <span class="sr-only">Toggle navigation</span>
                    <span class="icon-bar"></span>
                    <span class="icon-bar"></span>
                    <span class="icon-bar"></span>
                </button>
                <a asp-area="" asp-controller="Home" asp-action="Index" class="navbar-brand">EventSourceDemo</a>
            </div>
            <div class="navbar-collapse collapse">
                <ul class="nav navbar-nav">
                    <li>
                        <a asp-area="" asp-controller="Home" asp-action="Index">Home</a>
                    </li>
                    <li>
                        <a asp-area="" asp-controller="Home" asp-action="EventSource">Direct Event Stream</a>
                    </li>
                </ul>
            </div>
        </div>
    </div>
    <div class="container body-content">
        @RenderBody()
        <hr/>
        <footer>
            <p>© 2019 - EventSource示例</p>
        </footer>
    </div>
    
    <environment names="Development">
        <script src="~/lib/jquery/dist/jquery.js"></script>
        <script src="~/lib/bootstrap/dist/js/bootstrap.js"></script>
        <script src="~/js/site.js" asp-append-version="true"></script>
    </environment>
    <environment names="Staging,Production">
        <script src="https://ajax.aspnetcdn.com/ajax/jquery/jquery-2.2.0.min.js" asp-fallback-src="~/lib/jquery/dist/jquery.min.js" asp-fallback-test="window.jQuery"> </script>
        <script src="https://ajax.aspnetcdn.com/ajax/bootstrap/3.3.6/bootstrap.min.js" asp-fallback-src="~/lib/bootstrap/dist/js/bootstrap.min.js" asp-fallback-test="window.jQuery && window.jQuery.fn && window.jQuery.fn.modal"> </script>
        <script src="~/js/site.min.js" asp-append-version="true"></script>
    </environment>
    
    @RenderSection("scripts", required: false)
    </body>
    </html>

## \_ViewImports.cshtml ##

@using EventSourceDemo
    @addTagHelper *, Microsoft.AspNetCore.Mvc.TagHelpers

## \_ViewStart.cshtml ##

@{
        Layout = "_Layout";
    }

## Program.cs ##

using System;
    using System.Collections.Generic;
    using System.IO;
    using System.Linq;
    using System.Threading.Tasks;
    using Microsoft.AspNetCore;
    using Microsoft.AspNetCore.Hosting;
    using Microsoft.Extensions.Configuration;
    using Microsoft.Extensions.Logging;
    
    namespace coreEventSource
    { 
        public class Program
        { 
            public static void Main(string[] args)
            { 
                CreateWebHostBuilder(args).Build().Run();
            }
    
            public static IWebHostBuilder CreateWebHostBuilder(string[] args) =>
                WebHost.CreateDefaultBuilder(args)
                    .UseKestrel()
                    .UseContentRoot(Directory.GetCurrentDirectory())
                    .UseStartup<Startup>();
        }
    }

## Startup.cs ##

using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Threading.Tasks;
    using EventSourceDemo.Services;
    using Microsoft.AspNetCore.Builder;
    using Microsoft.AspNetCore.Hosting;
    using Microsoft.AspNetCore.Http;
    using Microsoft.Extensions.Configuration;
    using Microsoft.Extensions.DependencyInjection;
    using Microsoft.Extensions.Logging;
    
    namespace coreEventSource
    { 
        public class Startup
        { 
    
            public Startup(IHostingEnvironment env)
            { 
                var builder = new ConfigurationBuilder()
                    .SetBasePath(env.ContentRootPath)
                    .AddJsonFile("appsettings.json", optional: true, reloadOnChange: true)
                    .AddEnvironmentVariables();
                Configuration = builder.Build();
            }
    
            public IConfigurationRoot Configuration {  get; }
    
    
            // This method gets called by the runtime. Use this method to add services to the container.
            // For more information on how to configure your application, visit https://go.microsoft.com/fwlink/?LinkID=398940
            public void ConfigureServices(IServiceCollection services)
            { 
                // Add framework services.
                services.AddMvc();
    
                // Add logging
                services.AddLogging();
    
                // Add the event coordination service
                services.AddSingleton<EventsService>();
            }
    
            // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
            public void Configure(IApplicationBuilder app, IHostingEnvironment env, ILoggerFactory loggerFactory)
            { 
                loggerFactory.AddConsole(Configuration.GetSection("Logging"));
                loggerFactory.AddDebug();
    
                if (env.IsDevelopment())
                { 
                    app.UseDeveloperExceptionPage();
                    //app.UseBrowserLink();
                }
                else
                { 
                    app.UseExceptionHandler("/Home/Error");
                }
    
                app.UseStaticFiles();
    
                app.UseMvc(routes =>
                { 
                    routes.MapRoute(
                        name: "short",
                        template: "{action=Index}/{id?}",
                        defaults: new {  controller = "Home" });
    
                    routes.MapRoute(
                        name: "default",
                        template: "{controller=Home}/{action=Index}/{id?}");
                });
    
    
                if (env.IsDevelopment())
                { 
                    app.UseDeveloperExceptionPage();
                }
    
                //app.Run(async (context) =>
                //{ 
                // await context.Response.WriteAsync("Hello World!");
                //});
            }
        }
    }

## `launchSettings.json` ##

{ 
      "iisSettings": { 
        "windowsAuthentication": false, 
        "anonymousAuthentication": true, 
        "iisExpress": { 
          "applicationUrl": "http://localhost:51261",
          "sslPort": 0
        }
      },
      "profiles": { 
        "IIS Express": { 
          "commandName": "IISExpress",
          "launchBrowser": true,
          "environmentVariables": { 
            "ASPNETCORE_ENVIRONMENT": "Development"
          }
        },
        "coreEventSource": { 
          "commandName": "Project",
          "launchBrowser": true,
          "applicationUrl": "http://localhost:5000",
          "environmentVariables": { 
            "ASPNETCORE_ENVIRONMENT": "Development"
          }
        }
      }
    }

## 如图: ##

![在这里插入图片描述][20190802125931955.png]

[20190725172142494.png]: /images/20210626/8c72bc4aed8c41648d1e95d05b4e5380.png
[20190802125931955.png]: /images/20210626/bf069857680f4861ba7eaa1f299bb48f.png