Add possibility for callee instance to return an observable

#238

Author: Johan 't Hart

src/netstandard/Tests/WampSharp.Tests.Wampv2/Integration/RpcProgressTests.cs

@@ -41,6 +41,57 @@ public async Task ProgressiveCallsCallerProgress()
             Assert.That(callback.Task.Result, Is.EqualTo(10));
         }
 
+        [Test]
+        public async Task ProgressiveCallsCallerProgressObservable()
+        {
+            WampPlayground playground = new WampPlayground();
+
+            CallerCallee dualChannel = await playground.GetCallerCalleeDualChannel();
+            IWampChannel calleeChannel = dualChannel.CalleeChannel;
+            IWampChannel callerChannel = dualChannel.CallerChannel;
+
+            var service = new LongOpObsService();
+            await calleeChannel.RealmProxy.Services.RegisterCallee(service);
+
+            MyCallback callback = new MyCallback();
+
+            callerChannel.RealmProxy.RpcCatalog.Invoke
+                (callback,
+                    new CallOptions() { ReceiveProgress = true },
+                    "com.myapp.longop",
+                    new object[] { 10, false });
+
+            Assert.That(service.State, Is.EqualTo(LongOpObsService.EState.Called));
+            Assert.That(callback.Task.Result, Is.EqualTo(-1));
+            CollectionAssert.AreEquivalent(Enumerable.Range(0, 10), callback.ProgressiveResults);
+            Assert.That(service.State, Is.EqualTo(LongOpObsService.EState.Completed));
+        }
+
+        [Test]
+        public async Task ProgressiveCallsCallerProgressCancelObservable()
+        {
+            WampPlayground playground = new WampPlayground();
+
+            CallerCallee dualChannel = await playground.GetCallerCalleeDualChannel();
+            IWampChannel calleeChannel = dualChannel.CalleeChannel;
+            IWampChannel callerChannel = dualChannel.CallerChannel;
+
+            var service = new LongOpObsService();
+            await calleeChannel.RealmProxy.Services.RegisterCallee(service);
+
+            MyCallback callback = new MyCallback();
+
+            var invocation = callerChannel.RealmProxy.RpcCatalog.Invoke
+                (callback,
+                    new CallOptions() { ReceiveProgress = true },
+                    "com.myapp.longop",
+                    new object[] { 10, false });
+
+            Assert.That(service.State, Is.EqualTo(LongOpObsService.EState.Called));
+            invocation.Cancel(new CancelOptions());
+            Assert.That(service.State, Is.EqualTo(LongOpObsService.EState.Cancelled));
+        }
+
         [Test]
         public async Task ProgressiveCallsCalleeProxyProgress()
         {
@@ -180,19 +231,35 @@ public interface ILongOpObsService
 
         public class LongOpObsService : ILongOpObsService
         {
-            public IObservable<int> LongOp(int n, bool endWithError) => Observable.Create<int>(async obs =>
+            public enum EState
+            {
+                Nothing,
+                Called,
+                Completed,
+                Cancelled
+            }
+
+            public EState State { get; set; } = EState.Nothing;
+
+            public IObservable<int> LongOp(int n, bool endWithError) => Observable.Create<int>(async (obs, ct) =>
             {
+                State = EState.Called;
+                ct.Register(() =>
+                {
+                    if (State == EState.Called)
+                        State = EState.Cancelled;
+                });
                 for (int i = 0; i < n; i++)
                 {
                     obs.OnNext(i);
-                    await Task.Delay(100);
+                    await Task.Delay(100, ct);
+                    ct.ThrowIfCancellationRequested();
                 }
+                State = EState.Completed;
                 if (endWithError)
                     obs.OnError(new WampException("wamp.error", "Something bad happened"));
                 else
                     obs.OnCompleted();
-
-                return Disposable.Empty;
             });
         }
 
@@ -206,7 +273,7 @@ public class MyCallback : IWampRawRpcOperationClientCallback
 
             public void Result<TMessage>(IWampFormatter<TMessage> formatter, ResultDetails details)
             {
-                throw new NotImplementedException();
+                mTask.SetResult(-1); // -1 indicates no final return value
             }
 
             public void Result<TMessage>(IWampFormatter<TMessage> formatter, ResultDetails details, TMessage[] arguments)

src/netstandard/WampSharp/WAMP2/V2/Rpc/Callee/LocalRpcOperation.cs

@@ -77,7 +77,7 @@ protected void CallResult(IWampRawRpcOperationRouterCallback caller, YieldOption
             {
                 caller.Result(ObjectFormatter, options, arguments, argumentKeywords);
             }
-            else if (!this.HasResult)
+            else if (!this.HasResult || arguments == null)
             {
                 caller.Result(ObjectFormatter, options);
             }
@@ -93,7 +93,7 @@ protected IEnumerable<object> UnpackParameters<TMessage>(IWampFormatter<TMessage
         {
             ArgumentUnpacker unpacker = new ArgumentUnpacker(Parameters);
 
-            IEnumerable<object> result = 
+            IEnumerable<object> result =
                 unpacker.UnpackParameters(formatter, arguments, argumentsKeywords);
 
             return result;

src/netstandard/WampSharp/WAMP2/V2/Rpc/Callee/Reflection/OperationExtractor.cs

@@ -57,7 +57,11 @@ protected IWampRpcOperation CreateRpcMethod(Func<object> instanceProvider, ICall
             string procedureUri =
                 interceptor.GetProcedureUri(method);
 
-            if (!typeof (Task).IsAssignableFrom(method.ReturnType))
+            if (method.ReturnType.IsGenericType && method.ReturnType.GetGenericTypeDefinition() == typeof(IObservable<>))
+            {
+                return CreateProgressiveObservableOperation(instanceProvider, method, procedureUri);
+            }
+            else if (!typeof (Task).IsAssignableFrom(method.ReturnType))
             {
                 MethodInfoValidation.ValidateSyncMethod(method);
                 return new SyncMethodInfoRpcOperation(instanceProvider, method, procedureUri);
@@ -97,5 +101,25 @@ private static IWampRpcOperation CreateProgressiveOperation(Func<object> instanc
 
             return operation;
         }
+
+        private static IWampRpcOperation CreateProgressiveObservableOperation(Func<object> instanceProvider, MethodInfo method, string procedureUri)
+        {
+            //return new ProgressiveObservableMethodInfoRpcOperation<returnType>
+            // (instance, method, procedureUri);
+
+            Type returnType = method.ReturnType.GetGenericArguments()[0];
+
+            Type operationType =
+                typeof(ProgressiveObservableMethodInfoRpcOperation<>)
+                    .MakeGenericType(returnType);
+
+            IWampRpcOperation operation =
+                (IWampRpcOperation)Activator.CreateInstance(operationType,
+                    instanceProvider,
+                    method,
+                    procedureUri);
+
+            return operation;
+        }
     }
-}
+}

src/netstandard/WampSharp/WAMP2/V2/Rpc/Callee/Reflection/ProgressiveObservableMethodInfoRpcOperation.cs

@@ -0,0 +1,38 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reflection;
+using System.Threading;
+using WampSharp.Core.Utilities;
+using WampSharp.Core.Serialization;
+using WampSharp.V2.Core.Contracts;
+
+namespace WampSharp.V2.Rpc
+{
+    public class ProgressiveObservableMethodInfoRpcOperation<T> : SyncMethodInfoRpcOperation
+    {
+        public ProgressiveObservableMethodInfoRpcOperation(Func<object> instanceProvider, MethodInfo method, string procedureName) :
+            base(instanceProvider, method, procedureName)
+        {
+        }
+
+        protected override IWampCancellableInvocation OnResult(IWampRawRpcOperationRouterCallback caller, object result, IDictionary<string, object> outputs)
+        {
+            var ctSource = new CancellationTokenSource();
+            ((IObservable<T>)result).Subscribe(
+                it => CallResult(caller, it, outputs, new YieldOptions { Progress = true }),
+                ex =>
+                {
+                    if (ex is WampException wampex)
+                        HandleException(caller, wampex);
+                    else
+                        HandleException(caller, ex);
+                },
+                // An observable does not emit any value when completing, so result without arguments
+                () => caller.Result(ObjectFormatter, new YieldOptions()),
+                ctSource.Token
+            );
+            return new CancellationTokenSourceInvocation(ctSource);
+        }
+    }
+}

src/netstandard/WampSharp/WAMP2/V2/Rpc/Callee/SyncLocalRpcOperation.cs

@@ -29,30 +29,46 @@ protected override IWampCancellableInvocation InnerInvoke<TMessage>(IWampRawRpcO
                                argumentsKeywords,
                                out IDictionary<string, object> outputs);
 
-                CallResult(caller, result, outputs);
+                return OnResult(caller, result, outputs);
             }
             catch (WampException ex)
             {
-                mLogger.ErrorFormat(ex, "An error occurred while calling {ProcedureUri}", this.Procedure);
-                IWampErrorCallback callback = new WampRpcErrorCallback(caller);
-                callback.Error(ex);
+                HandleException(caller, ex);
             }
             catch (Exception ex)
             {
-                WampException wampException = ConvertExceptionToRuntimeException(ex);
-                IWampErrorCallback callback = new WampRpcErrorCallback(caller);
-                callback.Error(wampException);
+                HandleException(caller, ex);
             }
 
             return null;
         }
 
+        protected void HandleException(IWampRawRpcOperationRouterCallback caller, WampException ex)
+        {
+            mLogger.ErrorFormat(ex, "An error occurred while calling {ProcedureUri}", this.Procedure);
+            IWampErrorCallback callback = new WampRpcErrorCallback(caller);
+            callback.Error(ex);
+        }
+
+        protected void HandleException(IWampRawRpcOperationRouterCallback caller, Exception ex)
+        {
+            WampException wampException = ConvertExceptionToRuntimeException(ex);
+            IWampErrorCallback callback = new WampRpcErrorCallback(caller);
+            callback.Error(wampException);
+        }
+
+        protected virtual IWampCancellableInvocation OnResult(IWampRawRpcOperationRouterCallback caller, object result, IDictionary<string, object> outputs)
+        {
+            CallResult(caller, result, outputs);
+            return null;
+        }
+
         protected void CallResult(IWampRawRpcOperationRouterCallback caller, object result, IDictionary<string, object> outputs, YieldOptions yieldOptions = null)
         {
             yieldOptions = yieldOptions ?? new YieldOptions();
             object[] resultArguments = GetResultArguments(result);
 
-            IDictionary<string, object> argumentKeywords = 
+            IDictionary<string, object> argumentKeywords =
                 GetResultArgumentKeywords(result, outputs);
 
             CallResult(caller,
@@ -69,4 +85,4 @@ protected virtual IDictionary<string, object> GetResultArgumentKeywords(object r
         protected abstract object InvokeSync<TMessage>
             (IWampRawRpcOperationRouterCallback caller, IWampFormatter<TMessage> formatter, InvocationDetails details, TMessage[] arguments, IDictionary<string, TMessage> argumentsKeywords, out IDictionary<string, object> outputs);
     }
-}
+}