Observable progressive caller/callee (#355)

Caller/Callee observable support

Co-authored-by: Johan 't Hart <[email protected]>

Author: darkl

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

@@ -1,6 +1,8 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Disposables;
+using System.Reactive.Linq;
 using System.Threading.Tasks;
 using NUnit.Framework;
 using WampSharp.Core.Serialization;
@@ -30,13 +32,65 @@ public async Task ProgressiveCallsCallerProgress()
             callerChannel.RealmProxy.RpcCatalog.Invoke
                 (callback,
                     new CallOptions() {ReceiveProgress = true},
-                    "com.myapp.longop",
+                    MyOperation.ProcedureUri,
                     new object[] {10});
 
-            callback.Task.Wait(2000);
+            int? result = await callback.Task;
 
             CollectionAssert.AreEquivalent(Enumerable.Range(0, 10), callback.ProgressiveResults);
-            Assert.That(callback.Task.Result, Is.EqualTo(10));
+            Assert.That(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 LongOpObservableService();
+            await calleeChannel.RealmProxy.Services.RegisterCallee(service);
+
+            MyCallback callback = new MyCallback();
+
+            callerChannel.RealmProxy.RpcCatalog.Invoke
+                (callback,
+                    new CallOptions() { ReceiveProgress = true },
+                    MyOperation.ProcedureUri,
+                    new object[] { 10 });
+
+            Assert.That(service.State, Is.EqualTo(LongOpObservableService.OperationState.Called));
+            int? result = await callback.Task;
+            Assert.That(result, Is.EqualTo(null));
+            CollectionAssert.AreEquivalent(Enumerable.Range(0, 10), callback.ProgressiveResults);
+            Assert.That(service.State, Is.EqualTo(LongOpObservableService.OperationState.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 LongOpObservableService();
+            await calleeChannel.RealmProxy.Services.RegisterCallee(service);
+
+            MyCallback callback = new MyCallback();
+
+            var invocation = callerChannel.RealmProxy.RpcCatalog.Invoke
+                (callback,
+                    new CallOptions() { ReceiveProgress = true },
+                    MyOperation.ProcedureUri,
+                    new object[] { 10 });
+
+            Assert.That(service.State, Is.EqualTo(LongOpObservableService.OperationState.Called));
+            invocation.Cancel(new CancelOptions());
+            Assert.That(service.State, Is.EqualTo(LongOpObservableService.OperationState.Cancelled));
         }
 
         [Test]
@@ -56,17 +110,56 @@ public async Task ProgressiveCallsCalleeProxyProgress()
             List<int> results = new List<int>();
             MyProgress<int> progress = new MyProgress<int>(i => results.Add(i));
 
-            Task<int> result = proxy.LongOp(10, progress);
-            result.Wait();
+            int result = await proxy.LongOp(10, progress);
 
             CollectionAssert.AreEquivalent(Enumerable.Range(0, 10), results);
 
-            Assert.That(result.Result, Is.EqualTo(10));
+            Assert.That(result, Is.EqualTo(10));
+        }
+
+        [Test]
+        public async Task ProgressiveCallsCalleeProxyObservable()
+        {
+            WampPlayground playground = new WampPlayground();
+
+            CallerCallee dualChannel = await playground.GetCallerCalleeDualChannel();
+            IWampChannel calleeChannel = dualChannel.CalleeChannel;
+            IWampChannel callerChannel = dualChannel.CallerChannel;
+
+            MyOperation myOperation = new MyOperation();
+
+            await calleeChannel.RealmProxy.RpcCatalog.Register(myOperation, new RegisterOptions());
+            ILongOpObservableService proxy = callerChannel.RealmProxy.Services.GetCalleeProxy<ILongOpObservableService>();
+
+            IObservable<int> proxyResult = proxy.LongOp(9); // it will emit one more than asked
+
+            IEnumerable<int> results = proxyResult.ToEnumerable(); 
+
+            CollectionAssert.AreEquivalent(Enumerable.Range(0, 10), results);
+        }
+
+        [Test]
+        public async Task ProgressiveCallsCalleeProxyObservableError()
+        {
+            WampPlayground playground = new WampPlayground();
+
+            CallerCallee dualChannel = await playground.GetCallerCalleeDualChannel();
+            IWampChannel calleeChannel = dualChannel.CalleeChannel;
+            IWampChannel callerChannel = dualChannel.CallerChannel;
+
+            MyOperation myOperation = new MyOperation {EndWithError = true};
+
+            await calleeChannel.RealmProxy.RpcCatalog.Register(myOperation, new RegisterOptions());
+            ILongOpObservableService proxy = callerChannel.RealmProxy.Services.GetCalleeProxy<ILongOpObservableService>();
+
+            Assert.Throws(typeof(WampException), () => proxy.LongOp(9).ToEnumerable().Count());
         }
 
         public class MyOperation : IWampRpcOperation
         {
-            public string Procedure => "com.myapp.longop";
+            public const string ProcedureUri = "com.myapp.longop";
+
+            public string Procedure => ProcedureUri;
 
             public IWampCancellableInvocation Invoke<TMessage>(IWampRawRpcOperationRouterCallback caller, IWampFormatter<TMessage> formatter, InvocationDetails details)
             {
@@ -87,23 +180,35 @@ public IWampCancellableInvocation Invoke<TMessage>(IWampRawRpcOperationRouterCal
                         new object[] {i});
                 }
 
-                caller.Result(WampObjectFormatter.Value,
-                    new YieldOptions(),
-                    new object[] {n});
+                if (EndWithError)
+                {
+                    caller.Error(WampObjectFormatter.Value,
+                                 new Dictionary<string, string>(),
+                                 "longop.error",
+                                 new object[] {"Something bad happened"});
+                }
+                else
+                {
+                    caller.Result(WampObjectFormatter.Value,
+                        new YieldOptions(),
+                        new object[] { n });
+                }
 
                 return null;
             }
 
+            public bool EndWithError { get; set; }
+
             public IWampCancellableInvocation Invoke<TMessage>(IWampRawRpcOperationRouterCallback caller, IWampFormatter<TMessage> formatter, InvocationDetails details,
-                TMessage[] arguments, IDictionary<string, TMessage> argumentsKeywords)
+                                                               TMessage[] arguments, IDictionary<string, TMessage> argumentsKeywords)
             {
                 return null;
             }
         }
 
         public interface ILongOpService
         {
-            [WampProcedure("com.myapp.longop")]
+            [WampProcedure(MyOperation.ProcedureUri)]
             [WampProgressiveResultProcedure]
             Task<int> LongOp(int n, IProgress<int> progress);
         }
@@ -122,17 +227,71 @@ public async Task<int> LongOp(int n, IProgress<int> progress)
             }
         }
 
+        public interface ILongOpObservableService
+        {
+            [WampProcedure(MyOperation.ProcedureUri)]
+            [WampProgressiveResultProcedure]
+            IObservable<int> LongOp(int n);
+        }
+
+        public class LongOpObservableService : ILongOpObservableService
+        {
+            public enum OperationState
+            {
+                Nothing,
+                Called,
+                Completed,
+                Cancelled
+            }
+
+            public OperationState State { get; set; } = OperationState.Nothing;
+
+            public IObservable<int> LongOp(int n) => Observable.Create<int>(async (observer, cancellationToken) =>
+            {
+                State = OperationState.Called;
+
+                cancellationToken.Register(() =>
+                {
+                    if (State == OperationState.Called)
+                    {
+                        State = OperationState.Cancelled;
+                    }
+                });
+
+                for (int i = 0; i < n; i++)
+                {
+                    observer.OnNext(i);
+                    await Task.Delay(100, cancellationToken);
+                    cancellationToken.ThrowIfCancellationRequested();
+                }
+
+                State = OperationState.Completed;
+
+                if (EndWithError)
+                {
+                    observer.OnError(new WampException("longop.error", "Something bad happened"));
+                }
+                else
+                {
+                    observer.OnCompleted();
+                }
+            });
+
+            public bool EndWithError { get; set; }
+        }
+
         public class MyCallback : IWampRawRpcOperationClientCallback
         {
-            private readonly TaskCompletionSource<int> mTask = new TaskCompletionSource<int>();
+            private readonly TaskCompletionSource<int?> mTask = new TaskCompletionSource<int?>();
 
             public List<int> ProgressiveResults { get; } = new List<int>();
 
-            public Task<int> Task => mTask.Task;
+            public Task<int?> Task => mTask.Task;
 
             public void Result<TMessage>(IWampFormatter<TMessage> formatter, ResultDetails details)
             {
-                throw new NotImplementedException();
+                mTask.SetResult(null);
+                // null indicates no final return value
             }
 
             public void Result<TMessage>(IWampFormatter<TMessage> formatter, ResultDetails details, TMessage[] arguments)
@@ -187,4 +346,4 @@ public void Report(T value)
             mAction(value);
         }
     }
-}
+}

src/netstandard/WampSharp/WAMP2/V2/Api/CalleeProxy/CalleeProxyInterceptor.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Reflection;
 using WampSharp.V2.Core.Contracts;
 using WampSharp.V2.Rpc;
@@ -37,7 +38,7 @@ public virtual CallOptions GetCallOptions(MethodInfo method)
 
         public virtual string GetProcedureUri(MethodInfo method)
         {
-            WampProcedureAttribute attribute = 
+            WampProcedureAttribute attribute =
                 method.GetCustomAttribute<WampProcedureAttribute>();
 
             if (attribute == null)
@@ -48,4 +49,4 @@ public virtual string GetProcedureUri(MethodInfo method)
             return attribute.Procedure;
         }
     }
-}
+}

src/netstandard/WampSharp/WAMP2/V2/Api/CalleeProxy/CalleeProxyInterceptorFactory.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Reactive;
 using System.Reflection;
 using System.Threading.Tasks;
 using WampSharp.V2.Rpc;
@@ -28,7 +29,14 @@ private static Type GetRelevantInterceptorType(MethodInfo method)
             Type genericArgument;
             Type interceptorType;
 
-            if (!typeof(Task).IsAssignableFrom(returnType))
+            if (returnType.IsGenericType && returnType.GetGenericTypeDefinition() == typeof(IObservable<>))
+            {
+                MethodInfoValidation.ValidateProgressiveObservableMethod(method);
+
+                genericArgument = returnType.GetGenericArguments()[0];
+                interceptorType = typeof(ObservableCalleeProxyInterceptor<>);
+            }
+            else if (!typeof(Task).IsAssignableFrom(returnType))
             {
                 MethodInfoValidation.ValidateSyncMethod(method);
 
@@ -55,4 +63,4 @@ private static Type GetRelevantInterceptorType(MethodInfo method)
             return closedGenericType;
         }
     }
-}
+}

src/netstandard/WampSharp/WAMP2/V2/Api/CalleeProxy/ObservableCalleeProxyInterceptor.cs

@@ -0,0 +1,34 @@
+using System.Reactive;
+using System.Reactive.Linq;
+using System.Reflection;
+
+namespace WampSharp.V2.CalleeProxy
+{
+    internal class ObservableCalleeProxyInterceptor<T> : CalleeProxyInterceptorBase<T>
+    {
+        public ObservableCalleeProxyInterceptor(MethodInfo method, IWampCalleeProxyInvocationHandler handler, ICalleeProxyInterceptor interceptor) : base(method, handler, interceptor)
+        {
+        }
+
+        // We can't use Observable.FromAsync, as there is no overload that receives IObserver, and therefore we can't convert it into
+        // an IProgress instance and forward it to as the IProgress argument of InvokeProgressiveAsync.
+        // However, Observable.FromAsync and Observable.Create both use TaskObservableExtensions.Subscribe, which
+        // eventually uses EmitTaskResult so this is not a big deal.
+        public override object Invoke(MethodInfo method, object[] arguments)
+        {
+            return Observable.Create<T>(async (observer, cancellationToken) =>
+                                        {
+                                            T last = await Handler.InvokeProgressiveAsync
+                                                (Interceptor, method, Extractor, arguments, observer.ToProgress(),
+                                                 cancellationToken);
+
+                                            if (last != null)
+                                            {
+                                                observer.OnNext(last);
+                                            }
+
+                                            observer.OnCompleted();
+                                        });
+        }
+    }
+}

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

@@ -40,7 +40,7 @@ protected override IWampCancellableInvocation InnerInvoke<TMessage>(IWampRawRpcO
 
             if (SupportsCancellation)
             {
-                result = new CancellationTokenSourceInvocation(task ,cancellationTokenSource);
+                result = new CancellationTokenSourceInvocation(cancellationTokenSource);
             }
 
             return result;

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

@@ -1,18 +1,14 @@
 using System.Threading;
-using System.Threading.Tasks;
 using WampSharp.V2.Core.Contracts;
 
 namespace WampSharp.V2.Rpc
 {
     internal class CancellationTokenSourceInvocation : IWampCancellableInvocation
     {
-        private readonly Task mTask;
         private readonly CancellationTokenSource mCancellationTokenSource;
 
-        public CancellationTokenSourceInvocation(Task task,
-                                                 CancellationTokenSource cancellationTokenSource)
+        public CancellationTokenSourceInvocation(CancellationTokenSource cancellationTokenSource)
         {
-            mTask = task;
             mCancellationTokenSource = cancellationTokenSource;
         }
 

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

@@ -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;