From ca81db02f142a1c0a817d2f772ff3fcdabcb491f Mon Sep 17 00:00:00 2001 From: Elad Zelingher Date: Thu, 8 Aug 2019 12:05:31 -0400 Subject: [PATCH 1/7] Merge from async-rx --- .../WampSharp.RawSocket/RawSocket/RawSocketClientConnection.cs | 1 - .../Integration/PatternBasedPubSubTests.cs | 1 - .../WampSharp.Tests.Wampv2/Integration/PatternRpcTests.cs | 1 - .../Integration/PubSubReflectionTests.cs | 1 - .../WampSharp.Tests.Wampv2/Integration/PubSubRetainTests.cs | 1 - .../Integration/PubSubSubjectTupleTests.cs | 1 - .../Tests/WampSharp.Tests.Wampv2/Integration/SharedRpcTests.cs | 1 - .../Connections/AsyncConnection/AsyncWampConnection.cs | 1 - .../WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs | 1 - .../WAMP2/V2/Api/DelegatePubSub/WampSubscriberRegistrar.cs | 1 - src/net45/WampSharp/WAMP2/V2/Api/IWampRealmServiceProvider.cs | 1 - src/net45/WampSharp/WAMP2/V2/Api/Rx/WampClientSubject.cs | 1 - .../WAMP2/V2/Client/PubSub/Interfaces/IWampTopicProxy.cs | 3 --- .../V2/Client/PubSub/Interfaces/IWampTopicSubscriptionProxy.cs | 3 --- src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampSubscriber.cs | 1 - src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampTopicProxy.cs | 3 --- .../Rpc/Interfaces/IWampRpcOperationRegistrationProxy.cs | 3 --- src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampCallee.cs | 1 - .../WAMP2/V2/Client/Rpc/WampRpcOperationCatalogProxy.cs | 3 --- .../WampSharp/WAMP2/V2/MetaApi/Client/MetaApiEventsBase.cs | 1 - .../WampSharp/WAMP2/V2/MetaApi/Client/RegistrationEvents.cs | 1 - src/net45/WampSharp/WAMP2/V2/MetaApi/Client/SessionEvents.cs | 3 --- .../WampSharp/WAMP2/V2/MetaApi/Client/SubscriptionEvents.cs | 3 --- 23 files changed, 37 deletions(-) diff --git a/src/net45/Extensions/WampSharp.RawSocket/RawSocket/RawSocketClientConnection.cs b/src/net45/Extensions/WampSharp.RawSocket/RawSocket/RawSocketClientConnection.cs index 6f097339f..dae6f47d5 100644 --- a/src/net45/Extensions/WampSharp.RawSocket/RawSocket/RawSocketClientConnection.cs +++ b/src/net45/Extensions/WampSharp.RawSocket/RawSocket/RawSocketClientConnection.cs @@ -3,7 +3,6 @@ using System.Net.Security; using System.Net.Sockets; using System.Threading.Tasks; -using SystemEx; using Microsoft.IO; using WampSharp.Core.Listener; using WampSharp.Core.Message; diff --git a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PatternBasedPubSubTests.cs b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PatternBasedPubSubTests.cs index 7fd688e64..9dda58b20 100644 --- a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PatternBasedPubSubTests.cs +++ b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PatternBasedPubSubTests.cs @@ -2,7 +2,6 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; -using SystemEx; using NUnit.Framework; using WampSharp.Tests.Wampv2.TestHelpers.Integration; using WampSharp.V2; diff --git a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PatternRpcTests.cs b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PatternRpcTests.cs index 2490b6ed6..4e4ce3d29 100644 --- a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PatternRpcTests.cs +++ b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PatternRpcTests.cs @@ -1,7 +1,6 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; -using SystemEx; using NUnit.Framework; using WampSharp.Core.Serialization; using WampSharp.Tests.Wampv2.TestHelpers.Integration; diff --git a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubReflectionTests.cs b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubReflectionTests.cs index 520fa11e0..63861b79f 100644 --- a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubReflectionTests.cs +++ b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubReflectionTests.cs @@ -2,7 +2,6 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; -using SystemEx; using Newtonsoft.Json; using NUnit.Framework; using WampSharp.Core.Serialization; diff --git a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubRetainTests.cs b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubRetainTests.cs index e8823be27..8d2628e77 100644 --- a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubRetainTests.cs +++ b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubRetainTests.cs @@ -1,7 +1,6 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; -using SystemEx; using NUnit.Framework; using WampSharp.Tests.Wampv2.TestHelpers.Integration; using WampSharp.V2; diff --git a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubSubjectTupleTests.cs b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubSubjectTupleTests.cs index ef67f5025..32f133b3d 100644 --- a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubSubjectTupleTests.cs +++ b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/PubSubSubjectTupleTests.cs @@ -3,7 +3,6 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; -using SystemEx; using Newtonsoft.Json; using NUnit.Framework; using WampSharp.Core.Serialization; diff --git a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/SharedRpcTests.cs b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/SharedRpcTests.cs index 19046f93c..3a9898334 100644 --- a/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/SharedRpcTests.cs +++ b/src/net45/Tests/WampSharp.Tests.Wampv2/Integration/SharedRpcTests.cs @@ -2,7 +2,6 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; -using SystemEx; using NUnit.Framework; using WampSharp.Core.Serialization; using WampSharp.Tests.Wampv2.TestHelpers.Integration; diff --git a/src/net45/WampSharp/Core/Listener/Connections/AsyncConnection/AsyncWampConnection.cs b/src/net45/WampSharp/Core/Listener/Connections/AsyncConnection/AsyncWampConnection.cs index f73632ab8..416b40e94 100644 --- a/src/net45/WampSharp/Core/Listener/Connections/AsyncConnection/AsyncWampConnection.cs +++ b/src/net45/WampSharp/Core/Listener/Connections/AsyncConnection/AsyncWampConnection.cs @@ -1,7 +1,6 @@ using System; using System.Threading; using System.Threading.Tasks; -using SystemEx; using WampSharp.Core.Message; using WampSharp.Logging; diff --git a/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs b/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs index 5a8013ab9..cb7324ed3 100644 --- a/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs +++ b/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs @@ -2,7 +2,6 @@ using System.Collections.Generic; using System.Reactive.Subjects; using System.Threading.Tasks; -using SystemEx; using WampSharp.V2.CalleeProxy; using WampSharp.V2.Client; using WampSharp.V2.DelegatePubSub; diff --git a/src/net45/WampSharp/WAMP2/V2/Api/DelegatePubSub/WampSubscriberRegistrar.cs b/src/net45/WampSharp/WAMP2/V2/Api/DelegatePubSub/WampSubscriberRegistrar.cs index f1610a80f..c6786d464 100644 --- a/src/net45/WampSharp/WAMP2/V2/Api/DelegatePubSub/WampSubscriberRegistrar.cs +++ b/src/net45/WampSharp/WAMP2/V2/Api/DelegatePubSub/WampSubscriberRegistrar.cs @@ -2,7 +2,6 @@ using System.Collections.Generic; using System.Reflection; using System.Threading.Tasks; -using SystemEx; using WampSharp.V2.Client; using WampSharp.V2.Core.Contracts; using WampSharp.V2.PubSub; diff --git a/src/net45/WampSharp/WAMP2/V2/Api/IWampRealmServiceProvider.cs b/src/net45/WampSharp/WAMP2/V2/Api/IWampRealmServiceProvider.cs index d1c3df367..df286680c 100644 --- a/src/net45/WampSharp/WAMP2/V2/Api/IWampRealmServiceProvider.cs +++ b/src/net45/WampSharp/WAMP2/V2/Api/IWampRealmServiceProvider.cs @@ -1,7 +1,6 @@ using System; using System.Reactive.Subjects; using System.Threading.Tasks; -using SystemEx; using WampSharp.V2.PubSub; using WampSharp.V2.Rpc; diff --git a/src/net45/WampSharp/WAMP2/V2/Api/Rx/WampClientSubject.cs b/src/net45/WampSharp/WAMP2/V2/Api/Rx/WampClientSubject.cs index 5732bea59..96b2b978c 100644 --- a/src/net45/WampSharp/WAMP2/V2/Api/Rx/WampClientSubject.cs +++ b/src/net45/WampSharp/WAMP2/V2/Api/Rx/WampClientSubject.cs @@ -2,7 +2,6 @@ using System.Collections.Generic; using System.Reactive.Linq; using System.Threading.Tasks; -using SystemEx; using WampSharp.Core.Listener; using WampSharp.V2.Client; using WampSharp.V2.Core.Contracts; diff --git a/src/net45/WampSharp/WAMP2/V2/Client/PubSub/Interfaces/IWampTopicProxy.cs b/src/net45/WampSharp/WAMP2/V2/Client/PubSub/Interfaces/IWampTopicProxy.cs index 6be70607e..a1ae48645 100644 --- a/src/net45/WampSharp/WAMP2/V2/Client/PubSub/Interfaces/IWampTopicProxy.cs +++ b/src/net45/WampSharp/WAMP2/V2/Client/PubSub/Interfaces/IWampTopicProxy.cs @@ -1,9 +1,6 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; -#if !ASYNC_DISPOSABLE -using SystemEx; -#endif using WampSharp.V2.Core.Contracts; using WampSharp.V2.PubSub; diff --git a/src/net45/WampSharp/WAMP2/V2/Client/PubSub/Interfaces/IWampTopicSubscriptionProxy.cs b/src/net45/WampSharp/WAMP2/V2/Client/PubSub/Interfaces/IWampTopicSubscriptionProxy.cs index 71e25664a..50235f3ba 100644 --- a/src/net45/WampSharp/WAMP2/V2/Client/PubSub/Interfaces/IWampTopicSubscriptionProxy.cs +++ b/src/net45/WampSharp/WAMP2/V2/Client/PubSub/Interfaces/IWampTopicSubscriptionProxy.cs @@ -1,8 +1,5 @@ using System; using System.Threading.Tasks; -#if !ASYNC_DISPOSABLE -using SystemEx; -#endif using WampSharp.V2.Core.Contracts; using WampSharp.V2.PubSub; diff --git a/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampSubscriber.cs b/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampSubscriber.cs index d6783ba2e..58cdd441f 100644 --- a/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampSubscriber.cs +++ b/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampSubscriber.cs @@ -1,7 +1,6 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; -using SystemEx; using WampSharp.Core.Listener; using WampSharp.Core.Serialization; using WampSharp.Core.Utilities; diff --git a/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampTopicProxy.cs b/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampTopicProxy.cs index e759040fc..434e72354 100644 --- a/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampTopicProxy.cs +++ b/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampTopicProxy.cs @@ -1,9 +1,6 @@ using System; using System.Collections.Generic; using System.Threading.Tasks; -#if !ASYNC_DISPOSABLE -using SystemEx; -#endif using WampSharp.V2.Core.Contracts; using WampSharp.V2.PubSub; diff --git a/src/net45/WampSharp/WAMP2/V2/Client/Rpc/Interfaces/IWampRpcOperationRegistrationProxy.cs b/src/net45/WampSharp/WAMP2/V2/Client/Rpc/Interfaces/IWampRpcOperationRegistrationProxy.cs index ccaa72a4a..7d3a4356f 100644 --- a/src/net45/WampSharp/WAMP2/V2/Client/Rpc/Interfaces/IWampRpcOperationRegistrationProxy.cs +++ b/src/net45/WampSharp/WAMP2/V2/Client/Rpc/Interfaces/IWampRpcOperationRegistrationProxy.cs @@ -1,8 +1,5 @@ using System; using System.Threading.Tasks; -#if !ASYNC_DISPOSABLE -using SystemEx; -#endif using WampSharp.V2.Core.Contracts; using WampSharp.V2.Rpc; diff --git a/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampCallee.cs b/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampCallee.cs index 58992fff8..62e9cfc08 100644 --- a/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampCallee.cs +++ b/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampCallee.cs @@ -3,7 +3,6 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; -using SystemEx; using WampSharp.Core.Listener; using WampSharp.Core.Serialization; using WampSharp.Core.Utilities; diff --git a/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampRpcOperationCatalogProxy.cs b/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampRpcOperationCatalogProxy.cs index 648cf0a07..163c63496 100644 --- a/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampRpcOperationCatalogProxy.cs +++ b/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampRpcOperationCatalogProxy.cs @@ -2,9 +2,6 @@ using System.Collections.Generic; using System.Threading.Tasks; using WampSharp.Core.Serialization; -#if !ASYNC_DISPOSABLE -using SystemEx; -#endif using WampSharp.V2.Core.Contracts; using WampSharp.V2.Rpc; diff --git a/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/MetaApiEventsBase.cs b/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/MetaApiEventsBase.cs index 72af37430..3769caa3e 100644 --- a/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/MetaApiEventsBase.cs +++ b/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/MetaApiEventsBase.cs @@ -2,7 +2,6 @@ using System.Linq.Expressions; using System.Reflection; using System.Threading.Tasks; -using SystemEx; using WampSharp.Core.Utilities; using WampSharp.V2.Client; using WampSharp.V2.Core.Contracts; diff --git a/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/RegistrationEvents.cs b/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/RegistrationEvents.cs index 8eb2225e0..014cc76d4 100644 --- a/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/RegistrationEvents.cs +++ b/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/RegistrationEvents.cs @@ -1,6 +1,5 @@ using System; using System.Threading.Tasks; -using SystemEx; using WampSharp.V2.Client; namespace WampSharp.V2.MetaApi diff --git a/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/SessionEvents.cs b/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/SessionEvents.cs index 567920d2f..9b624b097 100644 --- a/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/SessionEvents.cs +++ b/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/SessionEvents.cs @@ -1,8 +1,5 @@ using System; using System.Threading.Tasks; -#if !ASYNC_DISPOSABLE -using SystemEx; -#endif using WampSharp.V2.Client; namespace WampSharp.V2.MetaApi diff --git a/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/SubscriptionEvents.cs b/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/SubscriptionEvents.cs index 23e1c434d..5633d436d 100644 --- a/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/SubscriptionEvents.cs +++ b/src/net45/WampSharp/WAMP2/V2/MetaApi/Client/SubscriptionEvents.cs @@ -1,8 +1,5 @@ using System; using System.Threading.Tasks; -#if !ASYNC_DISPOSABLE -using SystemEx; -#endif using WampSharp.V2.Client; namespace WampSharp.V2.MetaApi From 5423a2a2667b3b020d79b512c7ff5722afea9e87 Mon Sep 17 00:00:00 2001 From: Elad Zelingher Date: Fri, 2 Feb 2018 17:12:26 -0500 Subject: [PATCH 2/7] Adding async subjects implementation --- .../WAMP2/V2/Api/AsyncRx/IWampAsyncSubject.cs | 12 +++ .../AsyncRx/RawTopicClientAsyncSubscriber.cs | 33 +++++++ .../WAMP2/V2/Api/AsyncRx/WampAsyncSubject.cs | 54 +++++++++++ .../V2/Api/AsyncRx/WampClientAsyncSubject.cs | 96 +++++++++++++++++++ .../V2/Api/AsyncRx/WampTopicAsyncSubject.cs | 43 +++++++++ 5 files changed, 238 insertions(+) create mode 100644 src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/IWampAsyncSubject.cs create mode 100644 src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/RawTopicClientAsyncSubscriber.cs create mode 100644 src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampAsyncSubject.cs create mode 100644 src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampClientAsyncSubject.cs create mode 100644 src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTopicAsyncSubject.cs diff --git a/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/IWampAsyncSubject.cs b/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/IWampAsyncSubject.cs new file mode 100644 index 000000000..3789dc004 --- /dev/null +++ b/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/IWampAsyncSubject.cs @@ -0,0 +1,12 @@ +using System.Reactive.Subjects; + +namespace WampSharp.V2 +{ + /// + /// Represents a that publishes/receives messages + /// via a WAMP topic. + /// + public interface IWampAsyncSubject : IAsyncSubject + { + } +} \ No newline at end of file diff --git a/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/RawTopicClientAsyncSubscriber.cs b/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/RawTopicClientAsyncSubscriber.cs new file mode 100644 index 000000000..dad48835e --- /dev/null +++ b/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 mObserver; + + public RawTopicClientAsyncSubscriber(IAsyncObserver observer) + { + mObserver = observer; + } + + public void Event(IWampFormatter formatter, long publicationId, EventDetails details) + { + mObserver.OnNextAsync(new WampSerializedEvent(formatter, publicationId, details)); + } + + public void Event(IWampFormatter formatter, long publicationId, EventDetails details, TMessage[] arguments) + { + mObserver.OnNextAsync(new WampSerializedEvent(formatter, publicationId, details, arguments)); + } + + public void Event(IWampFormatter formatter, long publicationId, EventDetails details, TMessage[] arguments, IDictionary argumentsKeywords) + { + mObserver.OnNextAsync(new WampSerializedEvent(formatter, publicationId, details, arguments, argumentsKeywords)); + } + } +} \ No newline at end of file diff --git a/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampAsyncSubject.cs b/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampAsyncSubject.cs new file mode 100644 index 000000000..c2d47d076 --- /dev/null +++ b/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 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 argumentsKeywords); + + public abstract Task SubscribeAsync(IAsyncObserver observer); + } +} \ No newline at end of file diff --git a/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampClientAsyncSubject.cs b/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampClientAsyncSubject.cs new file mode 100644 index 000000000..53b76cdd8 --- /dev/null +++ b/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 mObservable; + + public WampClientAsyncSubject(IWampTopicProxy topic, IWampClientConnectionMonitor monitor) + { + mTopic = topic; + + mObservable = CreateObservable(topic, monitor); + } + + private static IAsyncObservable CreateObservable(IWampTopicProxy topic, IWampClientConnectionMonitor monitor) + { + IObservable connectionError = + Observable.FromEventPattern + (x => monitor.ConnectionError += x, + x => monitor.ConnectionError -= x) + .SelectMany(x => Observable.Throw(x.EventArgs.Exception)); + + IObservable connectionComplete = + Observable.FromEventPattern + (x => monitor.ConnectionBroken += x, + x => monitor.ConnectionBroken -= x) + .SelectMany(x => Observable.Throw(new WampConnectionBrokenException(x.EventArgs))); + + ClientAsyncObservable messages = new ClientAsyncObservable(topic); + + IObservable connectionNotifications = + Observable.Merge(connectionError, connectionComplete); + + IAsyncObservable asyncNotifications = + AsyncObservable.ToAsyncObservable(connectionNotifications); + + //IAsyncObservable result = + // AsyncObservable.Merge(asyncNotifications, messages); + + IAsyncObservable 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 argumentsKeywords) + { + return mTopic.Publish(options, arguments, argumentsKeywords); + } + + public override Task SubscribeAsync(IAsyncObserver observer) + { + return mObservable.SubscribeAsync(observer); + } + + private class ClientAsyncObservable : IAsyncObservable + { + private readonly IWampTopicProxy mTopic; + + public ClientAsyncObservable(IWampTopicProxy topic) + { + mTopic = topic; + } + + public Task SubscribeAsync(IAsyncObserver observer) + { + Task + task = mTopic.Subscribe(new RawTopicClientAsyncSubscriber(observer), new SubscribeOptions()); + + return task; + } + } + } +} \ No newline at end of file diff --git a/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTopicAsyncSubject.cs b/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTopicAsyncSubject.cs new file mode 100644 index 000000000..3e46d88a4 --- /dev/null +++ b/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 : IAsyncSubject + { + private readonly IWampAsyncSubject mSubject; + private readonly IAsyncObservable mObservable; + + public WampAsyncTopicSubject(IWampAsyncSubject subject) + { + mSubject = subject; + + mObservable = + subject.Where(x => x.Arguments != null && x.Arguments.Any()) + .Select(x => x.Arguments[0].Deserialize()); + } + + 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 SubscribeAsync(IAsyncObserver observer) + { + return mObservable.SubscribeAsync(observer); + } + } +} \ No newline at end of file From 490c6e00380973cafdfc5f041dc072fa3b56b4ec Mon Sep 17 00:00:00 2001 From: Elad Zelingher Date: Fri, 2 Feb 2018 17:13:10 -0500 Subject: [PATCH 3/7] Rename --- .../WampSharp/WAMP2/V2/Api/AsyncRx/WampTopicAsyncSubject.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTopicAsyncSubject.cs b/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTopicAsyncSubject.cs index 3e46d88a4..9856e7e0d 100644 --- a/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTopicAsyncSubject.cs +++ b/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTopicAsyncSubject.cs @@ -6,12 +6,12 @@ namespace WampSharp.V2 { - internal class WampAsyncTopicSubject : IAsyncSubject + internal class WampTopicAsyncSubject : IAsyncSubject { private readonly IWampAsyncSubject mSubject; private readonly IAsyncObservable mObservable; - public WampAsyncTopicSubject(IWampAsyncSubject subject) + public WampTopicAsyncSubject(IWampAsyncSubject subject) { mSubject = subject; From 121a547ccb1f3e345adc83920bc17d6d84678091 Mon Sep 17 00:00:00 2001 From: Elad Zelingher Date: Fri, 2 Feb 2018 17:18:47 -0500 Subject: [PATCH 4/7] Added async subjects overloads to IWampRealmServiceProvider --- .../Client/WampRealmProxyServiceProvider.cs | 23 ++++++++++++++++ .../WAMP2/V2/Api/IWampRealmServiceProvider.cs | 27 +++++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs b/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs index cb7324ed3..38b201eac 100644 --- a/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs +++ b/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs @@ -105,6 +105,24 @@ public IWampSubject GetSubject(string topicUri) return result; } + public IAsyncSubject GetAsyncSubject(string topicUri) + { + IWampAsyncSubject subject = GetAsyncSubject(topicUri); + + WampTopicAsyncSubject result = new WampTopicAsyncSubject(subject); + + return result; + } + + public IWampAsyncSubject GetAsyncSubject(string topicUri) + { + IWampTopicProxy topicProxy = mProxy.TopicContainer.GetTopicByUri(topicUri); + + WampClientAsyncSubject result = new WampClientAsyncSubject(topicProxy, mProxy.Monitor); + + return result; + } + public ISubject GetSubject(string topicUri, IWampEventValueTupleConverter converter) { IWampSubject subject = GetSubject(topicUri); @@ -114,6 +132,11 @@ public ISubject GetSubject(string topicUri, IWampEventValueTuple return result; } + public IAsyncSubject GetAsyncSubject(string topicUri, IWampEventValueTupleConverter tupleConverter) + { + throw new NotImplementedException(); + } + public IDisposable RegisterPublisher(object instance) { return RegisterPublisher(instance, new PublisherRegistrationInterceptor()); diff --git a/src/net45/WampSharp/WAMP2/V2/Api/IWampRealmServiceProvider.cs b/src/net45/WampSharp/WAMP2/V2/Api/IWampRealmServiceProvider.cs index df286680c..81355b304 100644 --- a/src/net45/WampSharp/WAMP2/V2/Api/IWampRealmServiceProvider.cs +++ b/src/net45/WampSharp/WAMP2/V2/Api/IWampRealmServiceProvider.cs @@ -102,6 +102,24 @@ public interface IWampRealmServiceProvider /// The requested subject. IWampSubject GetSubject(string topicUri); + /// + /// Gets a representing a + /// WAMP topic in the realm. + /// + /// The WAMP topic uri. + /// + /// The requested subject. + IAsyncSubject GetAsyncSubject(string topicUri); + + /// + /// Gets a representing a WAMP topic + /// in the realm. + /// + /// The WAMP topic uri. + /// The requested subject. + IWampAsyncSubject GetAsyncSubject(string topicUri); + + /// /// Registers an instance of a type having events decorated with /// to the realm. @@ -150,6 +168,15 @@ public interface IWampRealmServiceProvider /// The requested subject. ISubject GetSubject(string topicUri, IWampEventValueTupleConverter tupleConverter); + /// + /// Gets a representing a + /// WAMP topic in the realm. + /// + /// The WAMP topic uri. + /// An interface responsible for converting s into s + /// and vice versa + /// The requested subject. + IAsyncSubject GetAsyncSubject(string topicUri, IWampEventValueTupleConverter tupleConverter); #endif } From 2d98c8b31bbb4ef13011a1b0a5ad86ad4c3711d8 Mon Sep 17 00:00:00 2001 From: Elad Zelingher Date: Sun, 18 Feb 2018 15:57:53 -0500 Subject: [PATCH 5/7] Forgot implementing tuple support --- .../Api/AsyncRx/WampTupleTopicAsyncSubject.cs | 44 +++++++++++++++++++ .../Client/WampRealmProxyServiceProvider.cs | 6 ++- 2 files changed, 49 insertions(+), 1 deletion(-) create mode 100644 src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTupleTopicAsyncSubject.cs diff --git a/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTupleTopicAsyncSubject.cs b/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTupleTopicAsyncSubject.cs new file mode 100644 index 000000000..5b3c441a1 --- /dev/null +++ b/src/net45/WampSharp/WAMP2/V2/Api/AsyncRx/WampTupleTopicAsyncSubject.cs @@ -0,0 +1,44 @@ +using System; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Threading.Tasks; + +namespace WampSharp.V2 +{ + internal class WampTupleTopicAsyncSubject : IAsyncSubject + { + private readonly IWampAsyncSubject mSubject; + private readonly IWampEventValueTupleConverter mConverter; + private readonly IAsyncObservable mObservable; + + public WampTupleTopicAsyncSubject(IWampAsyncSubject subject, IWampEventValueTupleConverter converter) + { + mSubject = subject; + mConverter = converter; + + mObservable = + mSubject.Select(x => mConverter.ToTuple(x)); + } + + public Task OnNextAsync(TTuple value) + { + IWampEvent wampEvent = mConverter.ToEvent(value); + return mSubject.OnNextAsync(wampEvent); + } + + public Task OnErrorAsync(Exception error) + { + return mSubject.OnErrorAsync(error); + } + + public Task OnCompletedAsync() + { + return mSubject.OnCompletedAsync(); + } + + public Task SubscribeAsync(IAsyncObserver observer) + { + return mObservable.SubscribeAsync(observer); + } + } +} \ No newline at end of file diff --git a/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs b/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs index 38b201eac..51d9b3322 100644 --- a/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs +++ b/src/net45/WampSharp/WAMP2/V2/Api/Client/WampRealmProxyServiceProvider.cs @@ -134,7 +134,11 @@ public ISubject GetSubject(string topicUri, IWampEventValueTuple public IAsyncSubject GetAsyncSubject(string topicUri, IWampEventValueTupleConverter tupleConverter) { - throw new NotImplementedException(); + IWampAsyncSubject subject = GetAsyncSubject(topicUri); + + WampTupleTopicAsyncSubject result = new WampTupleTopicAsyncSubject(subject, tupleConverter); + + return result; } public IDisposable RegisterPublisher(object instance) From 2256f92dea259599c21d5cdf03b4f7d0c15bf2e0 Mon Sep 17 00:00:00 2001 From: Elad Zelingher Date: Thu, 8 Aug 2019 12:21:58 -0400 Subject: [PATCH 6/7] Adding missing using directives --- src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampSubscriber.cs | 1 + src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampCallee.cs | 1 + 2 files changed, 2 insertions(+) diff --git a/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampSubscriber.cs b/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampSubscriber.cs index 58cdd441f..2fef20cea 100644 --- a/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampSubscriber.cs +++ b/src/net45/WampSharp/WAMP2/V2/Client/PubSub/WampSubscriber.cs @@ -1,4 +1,5 @@ using System; +using SystemEx; using System.Collections.Generic; using System.Threading.Tasks; using WampSharp.Core.Listener; diff --git a/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampCallee.cs b/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampCallee.cs index 62e9cfc08..8527a14dc 100644 --- a/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampCallee.cs +++ b/src/net45/WampSharp/WAMP2/V2/Client/Rpc/WampCallee.cs @@ -1,4 +1,5 @@ using System; +using SystemEx; using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; From de3136aa616a76a5a050a386342b27d6a3c15dbb Mon Sep 17 00:00:00 2001 From: Elad Zelingher Date: Thu, 8 Aug 2019 13:01:49 -0400 Subject: [PATCH 7/7] Adding missing reference --- .../WAMP2/V2/Api/DelegatePubSub/WampSubscriberRegistrar.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/net45/WampSharp/WAMP2/V2/Api/DelegatePubSub/WampSubscriberRegistrar.cs b/src/net45/WampSharp/WAMP2/V2/Api/DelegatePubSub/WampSubscriberRegistrar.cs index c6786d464..f1610a80f 100644 --- a/src/net45/WampSharp/WAMP2/V2/Api/DelegatePubSub/WampSubscriberRegistrar.cs +++ b/src/net45/WampSharp/WAMP2/V2/Api/DelegatePubSub/WampSubscriberRegistrar.cs @@ -2,6 +2,7 @@ using System.Collections.Generic; using System.Reflection; using System.Threading.Tasks; +using SystemEx; using WampSharp.V2.Client; using WampSharp.V2.Core.Contracts; using WampSharp.V2.PubSub;