From 04fede0436d75c1f749c15217e5c7035d9638080 Mon Sep 17 00:00:00 2001
From: Pawel Kadluczka <moozzyk@gmail.com>
Date: Thu, 3 Nov 2016 21:58:10 -0700
Subject: [PATCH] Adding pluggable formatters

---
 .../FormatterResolver.cs                      | 44 +++++++++++++++++++
 .../PersistentConnectionLifeTimeManager.cs    |  9 +++-
 .../ProtobufWeatherStreamFormatter.cs         |  6 +--
 .../SocialWeatherEndPoint.cs                  | 18 ++++----
 samples/PersisitentConnection/Startup.cs      |  8 ++++
 .../PersisitentConnection/WeatherReport.cs    |  2 +
 .../wwwroot/weather.html                      |  3 +-
 7 files changed, 76 insertions(+), 14 deletions(-)
 create mode 100644 samples/PersisitentConnection/FormatterResolver.cs

diff --git a/samples/PersisitentConnection/FormatterResolver.cs b/samples/PersisitentConnection/FormatterResolver.cs
new file mode 100644
index 0000000000..f49d52cf43
--- /dev/null
+++ b/samples/PersisitentConnection/FormatterResolver.cs
@@ -0,0 +1,44 @@
+
+using System;
+using System.Collections.Generic;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace PersisitentConnection
+{
+    public class FormatterResolver
+    {
+        private IServiceProvider _serviceProvider;
+
+        private Dictionary<string, Dictionary<Type, Type>> _formatters
+            = new Dictionary<string, Dictionary<Type, Type>>();
+
+        public FormatterResolver(IServiceProvider serviceProvider)
+        {
+            _serviceProvider = serviceProvider;
+        }
+
+        public void AddFormatter<T, TFormatterType>(string formatType)
+            where TFormatterType : IStreamFormatter<T>
+        {
+            Dictionary<Type, Type> typeFormatters;
+            if (!_formatters.TryGetValue(formatType, out typeFormatters))
+            {
+                typeFormatters = _formatters[formatType] = new Dictionary<Type, Type>();
+            }
+            typeFormatters[typeof(T)] = typeof(TFormatterType);
+        }
+
+        public IStreamFormatter<T> GetFormatter<T>(string formatType)
+        {
+            Dictionary<Type, Type> typeFormatters;
+            Type typeFormatterType;
+            if (_formatters.TryGetValue(formatType, out typeFormatters) &&
+                typeFormatters.TryGetValue(typeof(T), out typeFormatterType))
+            {
+                return (IStreamFormatter<T>)_serviceProvider.GetRequiredService(typeFormatterType);
+            }
+
+            return null;
+        }
+    }
+}
diff --git a/samples/PersisitentConnection/PersistentConnectionLifeTimeManager.cs b/samples/PersisitentConnection/PersistentConnectionLifeTimeManager.cs
index 17514c5b2f..041a87f6e5 100644
--- a/samples/PersisitentConnection/PersistentConnectionLifeTimeManager.cs
+++ b/samples/PersisitentConnection/PersistentConnectionLifeTimeManager.cs
@@ -9,8 +9,14 @@ namespace PersisitentConnection
 {
     public class PersistentConnectionLifeTimeManager
     {
+        private readonly FormatterResolver _formatterResolver;
         private readonly ConnectionList _connectionList = new ConnectionList();
 
+        public PersistentConnectionLifeTimeManager(FormatterResolver formatterResolver)
+        {
+            _formatterResolver = formatterResolver;
+        }
+
         public void OnConnectedAsync(Connection connection)
         {
             _connectionList.Add(connection);
@@ -25,8 +31,7 @@ namespace PersisitentConnection
         {
             foreach (var connection in _connectionList)
             {
-//                var formatType = connection.Metadata.Get<string>("formatType");
-                var formatter = new JsonStreamFormatter<T>();
+                var formatter = _formatterResolver.GetFormatter<T>(connection.Metadata.Get<string>("formatType"));
                 await formatter.WriteAsync(data, connection.Channel.GetStream());
             }
         }
diff --git a/samples/PersisitentConnection/ProtobufWeatherStreamFormatter.cs b/samples/PersisitentConnection/ProtobufWeatherStreamFormatter.cs
index f9e27be53e..4a9c535252 100644
--- a/samples/PersisitentConnection/ProtobufWeatherStreamFormatter.cs
+++ b/samples/PersisitentConnection/ProtobufWeatherStreamFormatter.cs
@@ -4,14 +4,14 @@ using System.Threading.Tasks;
 
 namespace PersisitentConnection
 {
-    public class ProtobufWeatherStreamFormatter : IStreamFormatter<Weather>
+    public class ProtobufWeatherStreamFormatter : IStreamFormatter<WeatherReport>
     {
-        public Task<Weather> ReadAsync(Stream stream)
+        public Task<WeatherReport> ReadAsync(Stream stream)
         {
             throw new NotImplementedException();
         }
 
-        public Task WriteAsync(Weather value, Stream stream)
+        public Task WriteAsync(WeatherReport value, Stream stream)
         {
             throw new NotImplementedException();
         }
diff --git a/samples/PersisitentConnection/SocialWeatherEndPoint.cs b/samples/PersisitentConnection/SocialWeatherEndPoint.cs
index 43536e5d82..5b53dba9db 100644
--- a/samples/PersisitentConnection/SocialWeatherEndPoint.cs
+++ b/samples/PersisitentConnection/SocialWeatherEndPoint.cs
@@ -7,30 +7,32 @@ namespace PersisitentConnection
 {
     public class SocialWeatherEndPoint : EndPoint
     {
-        private readonly PersistentConnectionLifeTimeManager _lifetimeManager = new PersistentConnectionLifeTimeManager();
+        private readonly PersistentConnectionLifeTimeManager _lifetimeManager;
+        private readonly FormatterResolver _formatterResolver;
         private readonly ILogger<SocialWeatherEndPoint> _logger;
         private object _lockObj = new object();
         private WeatherReport _lastWeatherReport;
 
-        public SocialWeatherEndPoint(ILogger<SocialWeatherEndPoint> logger)
+        public SocialWeatherEndPoint(PersistentConnectionLifeTimeManager lifetimeManager,
+            FormatterResolver formatterResolver, ILogger<SocialWeatherEndPoint> logger)
         {
+            _lifetimeManager = lifetimeManager;
+            _formatterResolver = formatterResolver;
             _logger = logger;
         }
 
         public async override Task OnConnectedAsync(Connection connection)
         {
             _lifetimeManager.OnConnectedAsync(connection);
-            await DispatchMessagesAsync(connection);
+            await ProcessRequests(connection);
             _lifetimeManager.OnDisconnectedAsync(connection);
         }
 
-        public async Task DispatchMessagesAsync(Connection connection)
+        public async Task ProcessRequests(Connection connection)
         {
             var stream = connection.Channel.GetStream();
-            //var formatType = connection.Metadata.Get<string>("formatType");
-            //var formatterRegistry = _serviceProvider.GetRequiredService<FormatterRegistry>();
-            //var formatter = formatterRegistry.GetFormatter(formatType);
-            var formatter = new JsonStreamFormatter<WeatherReport>();
+            var formatter = _formatterResolver.GetFormatter<WeatherReport>(
+                connection.Metadata.Get<string>("formatType"));
 
             while (true)
             {
diff --git a/samples/PersisitentConnection/Startup.cs b/samples/PersisitentConnection/Startup.cs
index d3e9d3f410..775fd588a9 100644
--- a/samples/PersisitentConnection/Startup.cs
+++ b/samples/PersisitentConnection/Startup.cs
@@ -14,6 +14,10 @@ namespace PersisitentConnection
         {
             services.AddRouting();
             services.AddSingleton<SocialWeatherEndPoint>();
+            services.AddTransient<PersistentConnectionLifeTimeManager>();
+            services.AddSingleton(typeof(JsonStreamFormatter<>), typeof(JsonStreamFormatter<>));
+            services.AddSingleton<ProtobufWeatherStreamFormatter>();
+            services.AddSingleton<FormatterResolver>();
         }
 
         // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
@@ -29,6 +33,10 @@ namespace PersisitentConnection
             app.UseSockets(o => { o.MapEndpoint<SocialWeatherEndPoint>("/weather"); });
             app.UseStaticFiles();
 
+            var formatterResolver = app.ApplicationServices.GetRequiredService<FormatterResolver>();
+            formatterResolver.AddFormatter<WeatherReport, JsonStreamFormatter<WeatherReport>>("json");
+            formatterResolver.AddFormatter<WeatherReport, ProtobufWeatherStreamFormatter>("protobuf");
+
             app.Run(async (context) =>
             {
                 await context.Response.WriteAsync("Hello World!");
diff --git a/samples/PersisitentConnection/WeatherReport.cs b/samples/PersisitentConnection/WeatherReport.cs
index be24979446..7ed069af6b 100644
--- a/samples/PersisitentConnection/WeatherReport.cs
+++ b/samples/PersisitentConnection/WeatherReport.cs
@@ -14,5 +14,7 @@ namespace PersisitentConnection
         public long ReportTime { get; set; }
 
         public Weather Weather { get; set; }
+
+        public string ZipCode { get; set; }
     }
 }
diff --git a/samples/PersisitentConnection/wwwroot/weather.html b/samples/PersisitentConnection/wwwroot/weather.html
index 05128780fe..405131a8cd 100644
--- a/samples/PersisitentConnection/wwwroot/weather.html
+++ b/samples/PersisitentConnection/wwwroot/weather.html
@@ -43,7 +43,8 @@
                     JSON.stringify({
                         Temperature: '49',
                         Weather: 'Cloudy',
-                        ReportTime: Date.now()
+                        ReportTime: Date.now(),
+                        ZipCode: "98034"
                     }));
 
                 event.preventDefault();