Adding async subjects implementation

Author: darkl

src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/IWampAsyncSubject.cs

@@ -0,0 +1,12 @@
+using System.Reactive.Subjects;
+
+namespace WampSharp.V2
+{
+    /// <summary>
+    /// Represents a <see cref="IAsyncSubject{TSource,TResult}"/> that publishes/receives messages
+    /// via a WAMP topic.
+    /// </summary>
+    public interface IWampAsyncSubject : IAsyncSubject<IWampEvent, IWampSerializedEvent>
+    {
+    }
+}

src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/RawTopicClientAsyncSubscriber.cs

@@ -0,0 +1,33 @@
+using System;
+using System.Collections.Generic;
+using WampSharp.Core.Serialization;
+using WampSharp.V2.Core.Contracts;
+using WampSharp.V2.PubSub;
+
+namespace WampSharp.V2
+{
+    internal class RawTopicClientAsyncSubscriber : IWampRawTopicClientSubscriber
+    {
+        private readonly IAsyncObserver<IWampSerializedEvent> mObserver;
+
+        public RawTopicClientAsyncSubscriber(IAsyncObserver<IWampSerializedEvent> observer)
+        {
+            mObserver = observer;
+        }
+
+        public void Event<TMessage>(IWampFormatter<TMessage> formatter, long publicationId, EventDetails details)
+        {
+            mObserver.OnNextAsync(new WampSerializedEvent<TMessage>(formatter, publicationId, details));
+        }
+
+        public void Event<TMessage>(IWampFormatter<TMessage> formatter, long publicationId, EventDetails details, TMessage[] arguments)
+        {
+            mObserver.OnNextAsync(new WampSerializedEvent<TMessage>(formatter, publicationId, details, arguments));
+        }
+
+        public void Event<TMessage>(IWampFormatter<TMessage> formatter, long publicationId, EventDetails details, TMessage[] arguments, IDictionary<string, TMessage> argumentsKeywords)
+        {
+            mObserver.OnNextAsync(new WampSerializedEvent<TMessage>(formatter, publicationId, details, arguments, argumentsKeywords));
+        }
+    }
+}

src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampAsyncSubject.cs

@@ -0,0 +1,54 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using WampSharp.V2.Core.Contracts;
+
+namespace WampSharp.V2
+{
+    internal abstract class WampAsyncSubject : IWampAsyncSubject
+    {
+        private static readonly PublishOptions EmptyOptions  =
+            new PublishOptions();
+
+        public virtual Task OnNextAsync(IWampEvent value)
+        {
+            PublishOptions options = value.Options ?? EmptyOptions;
+            object[] arguments = value.Arguments;
+            IDictionary<string, object> argumentsKeywords = value.ArgumentsKeywords;
+
+            Task result;
+
+            if (argumentsKeywords != null)
+            {
+                result = Publish(options, arguments, argumentsKeywords);
+            }
+            else if (arguments != null)
+            {
+                result = Publish(options, arguments);
+            }
+            else
+            {
+                result = Publish(options);
+            }
+
+            return result;
+        }
+
+        public virtual Task OnErrorAsync(Exception error)
+        {
+            throw new NotImplementedException();
+        }
+
+        public virtual Task OnCompletedAsync()
+        {
+            throw new NotImplementedException();
+        }
+
+        protected abstract Task Publish(PublishOptions options);
+        protected abstract Task Publish(PublishOptions options, object[] arguments);
+        protected abstract Task Publish(PublishOptions options, object[] arguments,
+            IDictionary<string, object> argumentsKeywords);
+
+        public abstract Task<IAsyncDisposable> SubscribeAsync(IAsyncObserver<IWampSerializedEvent> observer);
+    }
+}

src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampClientAsyncSubject.cs

@@ -0,0 +1,96 @@
+using System;
+using System.Collections.Generic;
+using System.Reactive.Linq;
+using System.Threading.Tasks;
+using WampSharp.Core.Listener;
+using WampSharp.V2.Client;
+using WampSharp.V2.Core.Contracts;
+using WampSharp.V2.Realm;
+
+namespace WampSharp.V2
+{
+    internal class WampClientAsyncSubject : WampAsyncSubject
+    {
+        private readonly IWampTopicProxy mTopic;
+        private readonly IAsyncObservable<IWampSerializedEvent> mObservable;
+
+        public WampClientAsyncSubject(IWampTopicProxy topic, IWampClientConnectionMonitor monitor)
+        {
+            mTopic = topic;
+
+            mObservable = CreateObservable(topic, monitor);
+        }
+
+        private static IAsyncObservable<IWampSerializedEvent> CreateObservable(IWampTopicProxy topic, IWampClientConnectionMonitor monitor)
+        {
+            IObservable<IWampSerializedEvent> connectionError =
+                Observable.FromEventPattern<WampConnectionErrorEventArgs>
+                    (x => monitor.ConnectionError += x,
+                     x => monitor.ConnectionError -= x)
+                          .SelectMany(x => Observable.Throw<IWampSerializedEvent>(x.EventArgs.Exception));
+
+            IObservable<IWampSerializedEvent> connectionComplete =
+                Observable.FromEventPattern<WampSessionCloseEventArgs>
+                    (x => monitor.ConnectionBroken += x,
+                     x => monitor.ConnectionBroken -= x)
+                          .SelectMany(x => Observable.Throw<IWampSerializedEvent>(new WampConnectionBrokenException(x.EventArgs)));
+
+            ClientAsyncObservable messages = new ClientAsyncObservable(topic);
+
+            IObservable<IWampSerializedEvent> connectionNotifications =
+                Observable.Merge(connectionError, connectionComplete);
+
+            IAsyncObservable<IWampSerializedEvent> asyncNotifications =
+                AsyncObservable.ToAsyncObservable(connectionNotifications);
+
+            //IAsyncObservable<IWampSerializedEvent> result = 
+            //    AsyncObservable.Merge(asyncNotifications, messages);
+
+            IAsyncObservable<IWampSerializedEvent> result =
+                new[] {asyncNotifications, messages}.ToObservable()
+                    .ToAsyncObservable()
+                    .Merge();
+
+            return result;
+        }
+
+        protected override Task Publish(PublishOptions options)
+        {
+            return mTopic.Publish(options);
+        }
+
+        protected override Task Publish(PublishOptions options, object[] arguments)
+        {
+            return mTopic.Publish(options, arguments);
+        }
+
+        protected override Task Publish(PublishOptions options, object[] arguments,
+            IDictionary<string, object> argumentsKeywords)
+        {
+            return mTopic.Publish(options, arguments, argumentsKeywords);
+        }
+
+        public override Task<IAsyncDisposable> SubscribeAsync(IAsyncObserver<IWampSerializedEvent> observer)
+        {
+            return mObservable.SubscribeAsync(observer);
+        }
+
+        private class ClientAsyncObservable : IAsyncObservable<IWampSerializedEvent>
+        {
+            private readonly IWampTopicProxy mTopic;
+
+            public ClientAsyncObservable(IWampTopicProxy topic)
+            {
+                mTopic = topic;
+            }
+
+            public Task<IAsyncDisposable> SubscribeAsync(IAsyncObserver<IWampSerializedEvent> observer)
+            {
+                Task<IAsyncDisposable>
+                    task = mTopic.Subscribe(new RawTopicClientAsyncSubscriber(observer), new SubscribeOptions());
+
+                return task;
+            }
+        }
+    }
+}

src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTopicAsyncSubject.cs

@@ -0,0 +1,43 @@
+using System;
+using System.Linq;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using System.Threading.Tasks;
+
+namespace WampSharp.V2
+{
+    internal class WampAsyncTopicSubject<T> : IAsyncSubject<T>
+    {
+        private readonly IWampAsyncSubject mSubject;
+        private readonly IAsyncObservable<T> mObservable;
+
+        public WampAsyncTopicSubject(IWampAsyncSubject subject)
+        {
+            mSubject = subject;
+
+            mObservable =
+                subject.Where(x => x.Arguments != null && x.Arguments.Any())
+                       .Select(x => x.Arguments[0].Deserialize<T>());
+        }
+
+        public Task OnNextAsync(T value)
+        {
+            return mSubject.OnNextAsync(new WampEvent {Arguments = new object[] {value}});
+        }
+
+        public Task OnErrorAsync(Exception error)
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task OnCompletedAsync()
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task<IAsyncDisposable> SubscribeAsync(IAsyncObserver<T> observer)
+        {
+            return mObservable.SubscribeAsync(observer);
+        }
+    }
+}