fix test case 'CanReconnectWithSameObservable'

Author: rose-a

src/GraphQL.Client/GraphQLHttpClient.cs

@@ -5,6 +5,7 @@
 using System.Linq;
 using System.Net.Http;
 using System.Net.Http.Headers;
+using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using GraphQL.Client.Abstractions;
@@ -97,34 +98,15 @@ public Task<GraphQLResponse<TResponse>> SendMutationAsync<TResponse>(GraphQLRequ
 
         /// <inheritdoc />
         public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TResponse>(GraphQLRequest request)
-        {
-            if (_disposed)
-                throw new ObjectDisposedException(nameof(GraphQLHttpClient));
-
-            var key = new Tuple<GraphQLRequest, Type>(request, typeof(TResponse));
-
-            if (_subscriptionStreams.ContainsKey(key))
-                return (IObservable<GraphQLResponse<TResponse>>)_subscriptionStreams[key];
-
-            var observable = GraphQlHttpWebSocket.CreateSubscriptionStream<TResponse>(request);
-
-            _subscriptionStreams.TryAdd(key, observable);
-            return observable;
-        }
+            => CreateSubscriptionStream<TResponse>(request, null);
 
         /// <inheritdoc />
-        public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TResponse>(GraphQLRequest request, Action<Exception> exceptionHandler)
+        public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TResponse>(GraphQLRequest request, Action<Exception>? exceptionHandler)
         {
             if (_disposed)
                 throw new ObjectDisposedException(nameof(GraphQLHttpClient));
-
-            var key = new Tuple<GraphQLRequest, Type>(request, typeof(TResponse));
-
-            if (_subscriptionStreams.ContainsKey(key))
-                return (IObservable<GraphQLResponse<TResponse>>)_subscriptionStreams[key];
-
+            
             var observable = GraphQlHttpWebSocket.CreateSubscriptionStream<TResponse>(request, exceptionHandler);
-            _subscriptionStreams.TryAdd(key, observable);
             return observable;
         }
 

src/GraphQL.Client/Websocket/GraphQLHttpWebSocket.cs

@@ -94,7 +94,7 @@ public GraphQLHttpWebSocket(Uri webSocketUri, GraphQLHttpClient client)
         /// <param name="request">the <see cref="GraphQLRequest"/> to start the subscription</param>
         /// <param name="exceptionHandler">Optional: exception handler for handling exceptions within the receive pipeline</param>
         /// <returns>a <see cref="IObservable{TResponse}"/> which represents the subscription</returns>
-        public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TResponse>(GraphQLRequest request, Action<Exception> exceptionHandler = null) =>
+        public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TResponse>(GraphQLRequest request, Action<Exception>? exceptionHandler = null) =>
             Observable.Defer(() =>
                     Observable.Create<GraphQLResponse<TResponse>>(async observer =>
                     {
@@ -169,6 +169,7 @@ public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TRespons
                             observable.Subscribe(observer),
                             Disposable.Create(async () =>
                             {
+                                Debug.WriteLine($"disposing subscription {startRequest.Id}, websocket state is '{WebSocketState}'");
                                 // only try to send close request on open websocket
                                 if (WebSocketState != WebSocketState.Open)
                                     return;
@@ -252,10 +253,7 @@ public IObservable<GraphQLResponse<TResponse>> CreateSubscriptionStream<TRespons
                         return Observable.Empty<GraphQLResponse<TResponse>>();
                     }
                     return Observable.Return(t.Item1);
-                })
-                // transform to hot observable and auto-connect
-                .Publish().RefCount();
-
+                });
         /// <summary>
         /// Send a regular GraphQL request (query, mutation) via websocket
         /// </summary>

tests/GraphQL.Client.Tests.Common/Chat/Schema/IChat.cs

@@ -1,7 +1,10 @@
 using System;
 using System.Collections.Concurrent;
+using System.Diagnostics;
+using System.Reactive.Disposables;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
+using System.Threading;
 
 namespace GraphQL.Client.Tests.Common.Chat.Schema
 {
@@ -82,13 +85,20 @@ public MessageFrom Join(string userId)
         }
 
         public IObservable<Message> Messages(string user) =>
-            _messageStream
-                .Select(message =>
+            Observable.Create<Message>(observer =>
+            {
+                Debug.WriteLine($"creating messages stream for user '{user}' on thread {Thread.CurrentThread.ManagedThreadId}");
+                return new CompositeDisposable
                 {
-                    message.Sub = user;
-                    return message;
-                })
-                .AsObservable();
+                    _messageStream.Select(message =>
+                        {
+                            message.Sub = user;
+                            return message;
+                        })
+                        .Subscribe(observer),
+                    Disposable.Create(() => Debug.WriteLine($"disposing messages stream for user '{user}' on thread {Thread.CurrentThread.ManagedThreadId}"))
+                };
+            });
 
         public void AddError(Exception exception) => _messageStream.OnError(exception);
 

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

@@ -226,13 +226,13 @@ public async void CanReconnectWithSameObservable()
 
             const string message1 = "Hello World";
             Debug.WriteLine($"adding message {message1}");
-            var response = await ChatClient.AddMessageAsync(message1).ConfigureAwait(true);
+            var response = await ChatClient.AddMessageAsync(message1);
             response.Data.AddMessage.Content.Should().Be(message1);
             await observer.Should().PushAsync(2);
             observer.RecordedMessages.Last().Data.MessageAdded.Content.Should().Be(message1);
 
             const string message2 = "How are you?";
-            response = await ChatClient.AddMessageAsync(message2).ConfigureAwait(true);
+            response = await ChatClient.AddMessageAsync(message2);
             response.Data.AddMessage.Content.Should().Be(message2);
             await observer.Should().PushAsync(3);
             observer.RecordedMessages.Last().Data.MessageAdded.Content.Should().Be(message2);
@@ -247,7 +247,7 @@ public async void CanReconnectWithSameObservable()
             observer2.RecordedMessages.Last().Data.MessageAdded.Content.Should().Be(message2);
 
             const string message3 = "lorem ipsum dolor si amet";
-            response = await ChatClient.AddMessageAsync(message3).ConfigureAwait(true);
+            response = await ChatClient.AddMessageAsync(message3);
             response.Data.AddMessage.Content.Should().Be(message3);
             await observer2.Should().PushAsync(2);
             observer2.RecordedMessages.Last().Data.MessageAdded.Content.Should().Be(message3);