Added JSON RPC demo

samples/SocketsSample/EndPoints/Echo.cs

@@ -0,0 +1,15 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace SocketsSample
+{
+    public class Echo
+    {
+        public string Send(string message)
+        {
+            return message;
+        }
+    }
+}

samples/SocketsSample/EndPoints/JsonRpcEndpoint.cs

@@ -0,0 +1,153 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Reflection;
+using System.Threading.Tasks;
+using Channels;
+using Microsoft.AspNetCore.Sockets;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
+
+namespace SocketsSample
+{
+    // This end point implementation is used for framing JSON objects from the stream
+    public class JsonRpcEndpoint : EndPoint
+    {
+        private readonly Dictionary<string, Func<JObject, JObject>> _callbacks = new Dictionary<string, Func<JObject, JObject>>(StringComparer.OrdinalIgnoreCase);
+        private readonly ILogger<JsonRpcEndpoint> _logger;
+        private readonly IServiceProvider _serviceProvider;
+
+        public JsonRpcEndpoint(ILogger<JsonRpcEndpoint> logger, IServiceProvider serviceProvider)
+        {
+            // TODO: Discover end points
+            _logger = logger;
+            _serviceProvider = serviceProvider;
+
+            RegisterJsonRPCEndPoint(typeof(Echo));
+        }
+
+        public override async Task OnConnected(Connection connection)
+        {
+            // TODO: Dispatch from the caller
+            await Task.Yield();
+
+            // DO real async reads
+            var stream = connection.Channel.GetStream();
+            var reader = new JsonTextReader(new StreamReader(stream));
+            reader.SupportMultipleContent = true;
+
+            while (true)
+            {
+                while (!reader.Read())
+                {
+                    break;
+                }
+
+                JObject request;
+                try
+                {
+                    request = JObject.Load(reader);
+                }
+                catch (Exception)
+                {
+                    if (connection.Channel.Input.Reading.IsCompleted)
+                    {
+                        break;
+                    }
+
+                    throw;
+                }
+
+                if (_logger.IsEnabled(LogLevel.Debug))
+                {
+                    _logger.LogDebug("Received JSON RPC request: {request}", request);
+                }
+
+                JObject response = null;
+
+                Func<JObject, JObject> callback;
+                if (_callbacks.TryGetValue(request.Value<string>("method"), out callback))
+                {
+                    response = callback(request);
+                }
+                else
+                {
+                    // If there's no method then return a failed response for this request
+                    response = new JObject();
+                    response["id"] = request["id"];
+                    response["error"] = string.Format("Unknown method '{0}'", request.Value<string>("method"));
+                }
+
+                _logger.LogDebug("Sending JSON RPC response: {data}", response);
+
+                var writer = new JsonTextWriter(new StreamWriter(stream));
+                response.WriteTo(writer);
+                writer.Flush();
+            }
+        }
+
+        private void RegisterJsonRPCEndPoint(Type type)
+        {
+            var methods = new List<string>();
+
+            foreach (var m in type.GetTypeInfo().DeclaredMethods.Where(m => m.IsPublic))
+            {
+                var methodName = type.FullName + "." + m.Name;
+
+                methods.Add(methodName);
+
+                var parameters = m.GetParameters();
+
+                if (_callbacks.ContainsKey(methodName))
+                {
+                    throw new NotSupportedException(String.Format("Duplicate definitions of {0}. Overloading is not supported.", m.Name));
+                }
+
+                if (_logger.IsEnabled(LogLevel.Debug))
+                {
+                    _logger.LogDebug("RPC method '{methodName}' is bound", methodName);
+                }
+
+                _callbacks[methodName] = request =>
+                {
+                    var response = new JObject();
+                    response["id"] = request["id"];
+
+                    var scopeFactory = _serviceProvider.GetRequiredService<IServiceScopeFactory>();
+
+                    // Scope per call so that deps injected get disposed
+                    using (var scope = scopeFactory.CreateScope())
+                    {
+                        object value = scope.ServiceProvider.GetService(type) ?? Activator.CreateInstance(type);
+
+                        try
+                        {
+                            var args = request.Value<JArray>("params").Zip(parameters, (a, p) => a.ToObject(p.ParameterType))
+                                                                      .ToArray();
+
+                            var result = m.Invoke(value, args);
+
+                            if (result != null)
+                            {
+                                response["result"] = JToken.FromObject(result);
+                            }
+                        }
+                        catch (TargetInvocationException ex)
+                        {
+                            response["error"] = ex.InnerException.Message;
+                        }
+                        catch (Exception ex)
+                        {
+                            response["error"] = ex.Message;
+                        }
+                    }
+
+                    return response;
+                };
+            };
+        }
+    }
+}

samples/SocketsSample/Startup.cs

@@ -14,6 +14,7 @@ namespace SocketsSample
         {
             services.AddRouting();
 
+            services.AddSingleton<JsonRpcEndpoint>();
             services.AddSingleton<ChatEndPoint>();
         }
 
@@ -32,6 +33,7 @@ namespace SocketsSample
             app.UseSockets(d =>
             {
                 d.MapSocketEndpoint<ChatEndPoint>("/chat");
+                d.MapSocketEndpoint<JsonRpcEndpoint>("/jsonrpc");
             });
         }
     }

samples/SocketsSample/project.json

@@ -7,6 +7,7 @@
       "version": "1.0.0",
       "type": "platform"
     },
+    "Newtonsoft.Json": "9.0.1",
     "Microsoft.AspNetCore.Diagnostics": "1.1.0-*",
     "Microsoft.AspNetCore.StaticFiles": "1.1.0-*",
     "Microsoft.AspNetCore.Server.IISIntegration": "1.1.0-*",

samples/SocketsSample/wwwroot/index.html

@@ -5,11 +5,14 @@
     <title></title>
 </head>
 <body>
-    <h1>Chat sample</h1>
+    <h1>ASP.NET Sockets</h1>
+    <h2>Chat sample</h2>
     <ul>
         <li><a href="sse.html">Server Sent Events</a></li>
         <li><a href="polling.html">Long polling</a></li>
         <li><a href="ws.html">Web Sockets</a></li>
     </ul>
+    <h2>JSON RPC</h2>
+    <a href="rpc.html">JSON RPC</a>
 </body>
 </html>

samples/SocketsSample/wwwroot/rpc.html

@@ -0,0 +1,74 @@
+<!DOCTYPE html>
+<html>
+<head>
+    <meta charset="utf-8" />
+    <title></title>
+    <script>
+        function jsonRpc(url) {
+            var ws = new WebSocket(url);
+            var id = 0;
+            var calls = {};
+
+            ws.onopen = function () {
+                console.log('Opened!');
+            };
+
+            ws.onmessage = function (evt) {
+                var response = JSON.parse(evt.data);
+
+                var cb = calls[response.id];
+
+                delete calls[response.id];
+
+                if (response.error) {
+                    cb.error(response.error);
+                }
+                else {
+                    cb.success(response.result);
+                }
+            };
+
+            ws.onclose = function (evt) {
+                console.log('Closed!');
+            };
+
+            this.invoke = function (method, args) {
+                return new Promise((resolve, reject) => {
+                    calls[id] = { success: resolve, error: reject };
+                    ws.send(JSON.stringify({ method: method, params: args, id: id }));
+                    id++;
+                });
+            };
+        }
+
+
+        document.addEventListener('DOMContentLoaded', () => {
+            var rpc = new jsonRpc('ws://localhost:5000/jsonrpc/ws');
+
+            document.getElementById('sendmessage').addEventListener('click', () => {
+                let data = document.getElementById('data').value;
+
+                rpc.invoke('SocketsSample.Echo.Send', [data]).then(result => {
+                    var child = document.createElement('li');
+                    child.innerText = result;
+                    document.getElementById('messages').appendChild(child);
+                })
+                .catch(err => {
+                    var child = document.createElement('li');
+                    child.style.color = 'red';
+                    child.innerText = err;
+                    document.getElementById('messages').appendChild(child);
+                });
+            });
+        });
+
+    </script>
+</head>
+<body>
+    <h1>WebSockets</h1>
+    <input type="text" id="data" />
+    <input type="button" id="sendmessage" value="Send" />
+
+    <ul id="messages"></ul>
+</body>
+</html>