Fix concurrent subscriptions

root.props

@@ -17,7 +17,7 @@
 		<RepositoryType>git</RepositoryType>
 		<RepositoryUrl>https://github.com/graphql-dotnet/graphql-client.git</RepositoryUrl>
 		<TreatWarningsAsErrors>True</TreatWarningsAsErrors>
-		<Version>2.0.0-alpha.4.subscription-api.7</Version>
+		<Version>2.0.0-alpha.4.subscription-api.8</Version>
 		<WarningLevel>4</WarningLevel>
 	</PropertyGroup>
 

src/GraphQL.Client/Http/GraphQLHttpClient.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Net.Http;
 using System.Net.Http.Headers;
@@ -171,7 +172,7 @@ public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest requ
 
 			var observable = graphQlHttpWebSocket.CreateSubscriptionStream(request, Options, cancellationToken: _cancellationTokenSource.Token);
 
-			subscriptionStreams.Add(request, observable);
+			subscriptionStreams.TryAdd(request, observable);
 			return observable;
 		}
 
@@ -202,11 +203,11 @@ public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest requ
 				return subscriptionStreams[request];
 
 			var observable = graphQlHttpWebSocket.CreateSubscriptionStream(request, Options, exceptionHandler, _cancellationTokenSource.Token);
-			subscriptionStreams.Add(request, observable);
+			subscriptionStreams.TryAdd(request, observable);
 			return observable;
 		}
 
-		private Dictionary<GraphQLRequest, IObservable<GraphQLResponse>> subscriptionStreams = new Dictionary<GraphQLRequest, IObservable<GraphQLResponse>>();
+		private ConcurrentDictionary<GraphQLRequest, IObservable<GraphQLResponse>> subscriptionStreams = new ConcurrentDictionary<GraphQLRequest, IObservable<GraphQLResponse>>();
 
 		/// <summary>
 		/// Releases unmanaged resources

src/GraphQL.Client/Http/GraphQLHttpWebSocket.cs

@@ -81,6 +81,8 @@ await this.clientWebSocket.SendAsync(
 
 		public Task InitializeWebSocketTask { get; private set; } = Task.CompletedTask;
 
+		private object _initializeLock = new object();
+
 		#region Private Methods
 
 		private Task _backOff()
@@ -100,36 +102,39 @@ public Task InitializeWebSocket()
 			if(_disposed != null)
 				throw new OperationCanceledException();
 
-			// if an initialization task is already running, return that
-			if(InitializeWebSocketTask != null &&
-			   !InitializeWebSocketTask.IsFaulted &&
-			   !InitializeWebSocketTask.IsCompleted)
-				return InitializeWebSocketTask;
-
-			// if the websocket is open, return a completed task
-			if (clientWebSocket != null && clientWebSocket.State == WebSocketState.Open)
-				return Task.CompletedTask;
-
-			// else (re-)create websocket and connect
-			//_responseStreamConnection?.Dispose();
-			clientWebSocket?.Dispose();
-
-			// fix websocket not supported on win 7 using
-			// https://github.com/PingmanTools/System.Net.WebSockets.Client.Managed
-			clientWebSocket = SystemClientWebSocket.CreateClientWebSocket();
-			switch (clientWebSocket)
+			lock (_initializeLock)
 			{
-				case ClientWebSocket nativeWebSocket:
-					nativeWebSocket.Options.AddSubProtocol("graphql-ws");
-					break;
-				case System.Net.WebSockets.Managed.ClientWebSocket managedWebSocket:
-					managedWebSocket.Options.AddSubProtocol("graphql-ws");
-					break;
-				default:
-					throw new NotSupportedException($"unknown websocket type {clientWebSocket.GetType().Name}");
-			}
+				// if an initialization task is already running, return that
+				if(InitializeWebSocketTask != null &&
+				   !InitializeWebSocketTask.IsFaulted &&
+				   !InitializeWebSocketTask.IsCompleted)
+					return InitializeWebSocketTask;
+
+				// if the websocket is open, return a completed task
+				if (clientWebSocket != null && clientWebSocket.State == WebSocketState.Open)
+					return Task.CompletedTask;
+
+				// else (re-)create websocket and connect
+				//_responseStreamConnection?.Dispose();
+				clientWebSocket?.Dispose();
+
+				// fix websocket not supported on win 7 using
+				// https://github.com/PingmanTools/System.Net.WebSockets.Client.Managed
+				clientWebSocket = SystemClientWebSocket.CreateClientWebSocket();
+				switch (clientWebSocket)
+				{
+					case ClientWebSocket nativeWebSocket:
+						nativeWebSocket.Options.AddSubProtocol("graphql-ws");
+						break;
+					case System.Net.WebSockets.Managed.ClientWebSocket managedWebSocket:
+						managedWebSocket.Options.AddSubProtocol("graphql-ws");
+						break;
+					default:
+						throw new NotSupportedException($"unknown websocket type {clientWebSocket.GetType().Name}");
+				}
 
-			return InitializeWebSocketTask = _connectAsync(_cancellationTokenSource.Token);
+				return InitializeWebSocketTask = _connectAsync(_cancellationTokenSource.Token);
+			}
 		}
 
 		private IObservable<GraphQLWebSocketResponse> _createResponseStream()