Merge pull request #266 from graphql-dotnet/fix-260

Fix #260

Author: rose-a

.gitignore

@@ -4,3 +4,4 @@
 bin/
 obj/
 *.user
+nuget/

src/GraphQL.Client/Websocket/GraphQLHttpWebSocket.cs

@@ -4,7 +4,6 @@
 using System.Net.Http;
 using System.Net.WebSockets;
 using System.Reactive;
-using System.Reactive.Concurrency;
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
@@ -30,8 +29,6 @@ internal class GraphQLHttpWebSocket : IDisposable
         private readonly BehaviorSubject<GraphQLWebsocketConnectionState> _stateSubject =
             new BehaviorSubject<GraphQLWebsocketConnectionState>(GraphQLWebsocketConnectionState.Disconnected);
         private readonly IDisposable _requestSubscription;
-        private readonly EventLoopScheduler _receiveLoopScheduler = new EventLoopScheduler();
-        private readonly EventLoopScheduler _sendLoopScheduler = new EventLoopScheduler();
 
         private int _connectionAttempt = 0;
         private IConnectableObservable<WebsocketMessageWrapper> _incomingMessages;
@@ -80,12 +77,10 @@ public GraphQLHttpWebSocket(Uri webSocketUri, GraphQLHttpClient client)
             _client = client;
             _buffer = new ArraySegment<byte>(new byte[8192]);
             IncomingMessageStream = GetMessageStream();
-            _receiveLoopScheduler.Schedule(() =>
-                Debug.WriteLine($"receive loop scheduler thread id: {Thread.CurrentThread.ManagedThreadId}"));
 
             _requestSubscription = _requestSubject
-                .ObserveOn(_sendLoopScheduler)
-                .SelectMany(SendWebSocketRequestAsync)
+                .Select(SendWebSocketRequestAsync)
+                .Concat()
                 .Subscribe();
         }
 
@@ -436,7 +431,7 @@ private async Task ConnectAsync(CancellationToken token)
 
                 // create receiving observable
                 _incomingMessages = Observable
-                    .Defer(() => GetReceiveTask().ToObservable().ObserveOn(_receiveLoopScheduler))
+                    .Defer(() => GetReceiveTask().ToObservable())
                     .Repeat()
                     // complete sequence on OperationCanceledException, this is triggered by the cancellation token on disposal
                     .Catch<WebsocketMessageWrapper, OperationCanceledException>(exception => Observable.Empty<WebsocketMessageWrapper>())
@@ -489,13 +484,13 @@ private Task BackOff()
         }
 
         private IObservable<WebsocketMessageWrapper> GetMessageStream() =>
-            Observable.Using(() => new EventLoopScheduler(), scheduler =>
-                Observable.Create<WebsocketMessageWrapper>(async observer =>
+            Observable.Create<WebsocketMessageWrapper>(async observer =>
                 {
                     // make sure the websocket is connected
                     await InitializeWebSocket();
                     // subscribe observer to message stream
-                    var subscription = new CompositeDisposable(_incomingMessages.ObserveOn(scheduler).Subscribe(observer))
+                    var subscription = new CompositeDisposable(_incomingMessages
+                        .Subscribe(observer))
                     {
                         // register the observer's OnCompleted method with the cancellation token to complete the sequence on disposal
                         _internalCancellationTokenSource.Token.Register(observer.OnCompleted)
@@ -507,7 +502,7 @@ private IObservable<WebsocketMessageWrapper> GetMessageStream() =>
                     Debug.WriteLine($"new incoming message subscription {hashCode} created");
 
                     return subscription;
-                }));
+                });
 
         private Task<WebsocketMessageWrapper> _receiveAsyncTask = null;
         private readonly object _receiveTaskLocker = new object();
@@ -634,10 +629,7 @@ private async Task CompleteAsync()
             _exceptionSubject?.OnCompleted();
             _exceptionSubject?.Dispose();
             _internalCancellationTokenSource.Dispose();
-
-            _sendLoopScheduler?.Dispose();
-            _receiveLoopScheduler?.Dispose();
-
+            
             Debug.WriteLine("GraphQLHttpWebSocket disposed");
         }
 

tests/GraphQL.Integration.Tests/WebsocketTests/Base.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
 using System.Net.WebSockets;
@@ -257,8 +258,26 @@ public async void CanConnectTwoSubscriptionsSimultaneously()
             var observable2 = ChatClient.CreateSubscriptionStream<UserJoinedSubscriptionResult>(_subscriptionRequest2, callbackTester2.Invoke);
 
             Debug.WriteLine("subscribing...");
-            var messagesMonitor = observable1.Observe();
-            var joinedMonitor = observable2.Observe();
+            var blocker = new ManualResetEventSlim(false);
+            FluentTestObserver<GraphQLResponse<MessageAddedSubscriptionResult>> messagesMonitor = null;
+            FluentTestObserver<GraphQLResponse<UserJoinedSubscriptionResult>> joinedMonitor = null;
+
+            var tasks = new List<Task>
+            {
+                Task.Run(() =>
+                {
+                    blocker.Wait();
+                    messagesMonitor = observable1.Observe();
+                }),
+                Task.Run(() =>
+                {
+                    blocker.Wait();
+                    joinedMonitor = observable2.Observe();
+                })
+            };
+
+            blocker.Set();
+            await Task.WhenAll(tasks);
 
             await messagesMonitor.Should().PushAsync(1);
             messagesMonitor.RecordedMessages.Last().Data.MessageAdded.Content.Should().Be(InitialMessage.Content);