Handling custom protobuf types

samples/SocketsSample/EndPoints/ChatEndPoint.cs

@@ -1,6 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
+using System.Collections.Generic;
 using System.Text;
 using System.Threading.Tasks;
 using Channels;

samples/SocketsSample/EndPoints/HubEndpoint.cs

@@ -59,6 +59,7 @@ namespace SocketsSample
                     Arguments = args
                 };
 
+                // TODO: serialize once per format by providing a different stream?
                 foreach (var connection in _endPoint.Connections)
                 {
 

samples/SocketsSample/Hubs/Chat.cs

@@ -11,5 +11,10 @@ namespace SocketsSample.Hubs
         {
             Clients.All.Invoke("Send", message);
         }
+
+        public Person EchoPerson(Person p)
+        {
+            return p;
+        }
     }
 }

samples/SocketsSample/Hubs/Person.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+
+namespace SocketsSample.Hubs
+{
+    public class Person
+    {
+        public string Name { get; set; }
+        public long Age { get; set; }
+    }
+}

samples/SocketsSample/Protobuf/ProtobufInvocationAdapter.cs

@@ -2,17 +2,25 @@
 using System.IO;
 using System.Threading.Tasks;
 using Google.Protobuf;
+using Microsoft.Extensions.DependencyInjection;
 
 namespace SocketsSample.Protobuf
 {
     public class ProtobufInvocationAdapter : IInvocationAdapter
     {
+        IServiceProvider _serviceProvider;
+
+        public ProtobufInvocationAdapter(IServiceProvider serviceProvider)
+        {
+            _serviceProvider = serviceProvider;
+        }
+
         public async Task<InvocationDescriptor> CreateInvocationDescriptor(Stream stream, Func<string, Type[]> getParams)
         {
             return await Task.Run(() => CreateInvocationDescriptorInt(stream, getParams));
         }
 
-        private static Task<InvocationDescriptor> CreateInvocationDescriptorInt(Stream stream, Func<string, Type[]> getParams)
+        private Task<InvocationDescriptor> CreateInvocationDescriptorInt(Stream stream, Func<string, Type[]> getParams)
         {
             var inputStream = new CodedInputStream(stream, leaveOpen: true);
             var invocationHeader = new RpcInvocationHeader();
@@ -28,19 +36,22 @@ namespace SocketsSample.Protobuf
 
             for (var i = 0; i < argumentTypes.Length; i++)
             {
-                var value = new PrimitiveValue();
-                inputStream.ReadMessage(value);
                 if (typeof(int) == argumentTypes[i])
                 {
+                    var value = new PrimitiveValue();
+                    inputStream.ReadMessage(value);
                     invocationDescriptor.Arguments[i] = value.Int32Value;
                 }
                 else if (typeof(string) == argumentTypes[i])
                 {
+                    var value = new PrimitiveValue();
+                    inputStream.ReadMessage(value);
                     invocationDescriptor.Arguments[i] = value.StringValue;
                 }
                 else
                 {
-                    throw new InvalidOperationException();
+                    var serializer = _serviceProvider.GetRequiredService<ProtobufSerializer>();
+                    invocationDescriptor.Arguments[i] = serializer.GetValue(inputStream, argumentTypes[i]);
                 }
             }
 
@@ -65,7 +76,7 @@ namespace SocketsSample.Protobuf
 
             outputStream.WriteMessage(resultHeader);
 
-            if (resultHeader.Error == null && resultDescriptor.Result != null)
+            if (string.IsNullOrEmpty(resultHeader.Error) && resultDescriptor.Result != null)
             {
                 var result = resultDescriptor.Result;
 
@@ -77,6 +88,12 @@ namespace SocketsSample.Protobuf
                 {
                     outputStream.WriteMessage(new PrimitiveValue { StringValue = (string)result });
                 }
+                else
+                {
+                    var serializer = _serviceProvider.GetRequiredService<ProtobufSerializer>();
+                    var message = serializer.GetMessage(result);
+                    outputStream.WriteMessage(message);
+                }
             }
 
             outputStream.Flush();

samples/SocketsSample/Protobuf/RpcInvocation.cs

@@ -27,14 +27,16 @@ public static partial class RpcInvocationReflection {
           "IAEoBSJJChlScGNJbnZvY2F0aW9uUmVzdWx0SGVhZGVyEgoKAklkGAEgASgF",
           "EhEKCUhhc1Jlc3VsdBgCIAEoCBINCgVFcnJvchgDIAEoCSJHCg5QcmltaXRp",
           "dmVWYWx1ZRIUCgpJbnQzMlZhbHVlGAEgASgFSAASFQoLU3RyaW5nVmFsdWUY",
-          "AiABKAlIAEIICgZvbmVvZl9iBnByb3RvMw=="));
+          "AiABKAlIAEIICgZvbmVvZl8iKgoNUGVyc29uTWVzc2FnZRIMCgROYW1lGAEg",
+          "ASgJEgsKA0FnZRgCIAEoA2IGcHJvdG8z"));
     descriptor = pbr::FileDescriptor.FromGeneratedCode(descriptorData,
         new pbr::FileDescriptor[] { },
         new pbr::GeneratedClrTypeInfo(null, new pbr::GeneratedClrTypeInfo[] {
           new pbr::GeneratedClrTypeInfo(typeof(global::RpcMessageKind), global::RpcMessageKind.Parser, new[]{ "MessageKind" }, null, new[]{ typeof(global::RpcMessageKind.Types.Kind) }, null),
           new pbr::GeneratedClrTypeInfo(typeof(global::RpcInvocationHeader), global::RpcInvocationHeader.Parser, new[]{ "Name", "Id", "NumArgs" }, null, null, null),
           new pbr::GeneratedClrTypeInfo(typeof(global::RpcInvocationResultHeader), global::RpcInvocationResultHeader.Parser, new[]{ "Id", "HasResult", "Error" }, null, null, null),
-          new pbr::GeneratedClrTypeInfo(typeof(global::PrimitiveValue), global::PrimitiveValue.Parser, new[]{ "Int32Value", "StringValue" }, new[]{ "Oneof" }, null, null)
+          new pbr::GeneratedClrTypeInfo(typeof(global::PrimitiveValue), global::PrimitiveValue.Parser, new[]{ "Int32Value", "StringValue" }, new[]{ "Oneof" }, null, null),
+          new pbr::GeneratedClrTypeInfo(typeof(global::PersonMessage), global::PersonMessage.Parser, new[]{ "Name", "Age" }, null, null, null)
         }));
   }
   #endregion
@@ -692,6 +694,151 @@ public sealed partial class PrimitiveValue : pb::IMessage<PrimitiveValue> {
 
 }
 
+public sealed partial class PersonMessage : pb::IMessage<PersonMessage> {
+  private static readonly pb::MessageParser<PersonMessage> _parser = new pb::MessageParser<PersonMessage>(() => new PersonMessage());
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public static pb::MessageParser<PersonMessage> Parser { get { return _parser; } }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public static pbr::MessageDescriptor Descriptor {
+    get { return global::RpcInvocationReflection.Descriptor.MessageTypes[4]; }
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  pbr::MessageDescriptor pb::IMessage.Descriptor {
+    get { return Descriptor; }
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public PersonMessage() {
+    OnConstruction();
+  }
+
+  partial void OnConstruction();
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public PersonMessage(PersonMessage other) : this() {
+    name_ = other.name_;
+    age_ = other.age_;
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public PersonMessage Clone() {
+    return new PersonMessage(this);
+  }
+
+  /// <summary>Field number for the "Name" field.</summary>
+  public const int NameFieldNumber = 1;
+  private string name_ = "";
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public string Name {
+    get { return name_; }
+    set {
+      name_ = pb::ProtoPreconditions.CheckNotNull(value, "value");
+    }
+  }
+
+  /// <summary>Field number for the "Age" field.</summary>
+  public const int AgeFieldNumber = 2;
+  private long age_;
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public long Age {
+    get { return age_; }
+    set {
+      age_ = value;
+    }
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public override bool Equals(object other) {
+    return Equals(other as PersonMessage);
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public bool Equals(PersonMessage other) {
+    if (ReferenceEquals(other, null)) {
+      return false;
+    }
+    if (ReferenceEquals(other, this)) {
+      return true;
+    }
+    if (Name != other.Name) return false;
+    if (Age != other.Age) return false;
+    return true;
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public override int GetHashCode() {
+    int hash = 1;
+    if (Name.Length != 0) hash ^= Name.GetHashCode();
+    if (Age != 0L) hash ^= Age.GetHashCode();
+    return hash;
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public override string ToString() {
+    return pb::JsonFormatter.ToDiagnosticString(this);
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public void WriteTo(pb::CodedOutputStream output) {
+    if (Name.Length != 0) {
+      output.WriteRawTag(10);
+      output.WriteString(Name);
+    }
+    if (Age != 0L) {
+      output.WriteRawTag(16);
+      output.WriteInt64(Age);
+    }
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public int CalculateSize() {
+    int size = 0;
+    if (Name.Length != 0) {
+      size += 1 + pb::CodedOutputStream.ComputeStringSize(Name);
+    }
+    if (Age != 0L) {
+      size += 1 + pb::CodedOutputStream.ComputeInt64Size(Age);
+    }
+    return size;
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public void MergeFrom(PersonMessage other) {
+    if (other == null) {
+      return;
+    }
+    if (other.Name.Length != 0) {
+      Name = other.Name;
+    }
+    if (other.Age != 0L) {
+      Age = other.Age;
+    }
+  }
+
+  [global::System.Diagnostics.DebuggerNonUserCodeAttribute]
+  public void MergeFrom(pb::CodedInputStream input) {
+    uint tag;
+    while ((tag = input.ReadTag()) != 0) {
+      switch(tag) {
+        default:
+          input.SkipLastField();
+          break;
+        case 10: {
+          Name = input.ReadString();
+          break;
+        }
+        case 16: {
+          Age = input.ReadInt64();
+          break;
+        }
+      }
+    }
+  }
+
+}
+
 #endregion
 
 

samples/SocketsSample/Protobuf/RpcInvocation.proto

@@ -22,4 +22,9 @@ message PrimitiveValue {
         int32 Int32Value = 1;
         string StringValue = 2;
     }
+}
+
+message PersonMessage {
+  string Name = 1;
+  int64 Age = 2;
 }

samples/SocketsSample/ProtobufSerializer.cs

@@ -0,0 +1,33 @@
+using System;
+using Google.Protobuf;
+using SocketsSample.Hubs;
+
+namespace SocketsSample
+{
+    public class ProtobufSerializer
+    {
+        public object GetValue(CodedInputStream inputStream, Type type)
+        {
+            if (type == typeof(Person))
+            {
+                var value = new PersonMessage();
+                inputStream.ReadMessage(value);
+
+                return new Person { Name = value.Name, Age = value.Age };
+            }
+
+            throw new InvalidOperationException("(Deserialize) Unknown type.");
+        }
+
+        public IMessage GetMessage(object value)
+        {
+            Person person = value as Person;
+            if (person != null)
+            {
+                return new PersonMessage { Name = person.Name, Age = person.Age };
+            }
+
+            throw new InvalidOperationException("(Serialize) Unknown type.");
+        }
+    }
+}

samples/SocketsSample/Startup.cs

@@ -20,6 +20,7 @@ namespace SocketsSample
             services.AddSingleton<RpcEndpoint>();
             services.AddSingleton<ChatEndPoint>();
 
+            services.AddSingleton<ProtobufSerializer>();
             services.AddSingleton<SocketFormatters>();
         }
 
@@ -45,7 +46,7 @@ namespace SocketsSample
 
             app.UseFormatters(formatters=>
             {
-                formatters.AddInvocationAdapter("protobuf", new Protobuf.ProtobufInvocationAdapter());
+                formatters.AddInvocationAdapter("protobuf", new Protobuf.ProtobufInvocationAdapter(app.ApplicationServices));
                 formatters.AddInvocationAdapter("json", new JSonInvocationAdapter());
                 formatters.AddInvocationAdapter("line", new LineInvocationAdapter());
             });