Clean up subscription API

SubscriptionIntegrationTest.ConsoleClient/SubscriptionIntegrationTest.ConsoleClient.csproj

@@ -7,7 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="System.Reactive" Version="4.1.2" />
+    <PackageReference Include="System.Reactive" Version="4.1.3" />
   </ItemGroup>
 
   <ItemGroup>

benchmarks/GraphQL.Common.Benchmark/GraphQL.Common.Benchmark.csproj

@@ -1,12 +1,12 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
 	<PropertyGroup>
 		<OutputType>Exe</OutputType>
-		<TargetFramework>netcoreapp2.2</TargetFramework>
+		<TargetFramework>netcoreapp3.0</TargetFramework>
 	</PropertyGroup>
 
 	<ItemGroup>
-		<PackageReference Include="BenchmarkDotNet" Version="0.11.3" />
+		<PackageReference Include="BenchmarkDotNet" Version="0.11.4" />
 	</ItemGroup>
 
 	<ItemGroup>

samples/GraphQL.Client.Sample/Program.cs

@@ -1,15 +1,15 @@
 using System;
 using GraphQL.Client.Http;
+using GraphQL.Common.Request;
 
 namespace GraphQL.Client.Sample {
 
 	public class Program {
 
 		public static void Main(string[] args) {
 			using (var graphQLHttpClient = new GraphQLHttpClient("http://localhost:60341/graphql")) {
-				var subscriptionResult = graphQLHttpClient.SendSubscribeAsync(@"subscription { messageAdded{content}}").Result;
-				subscriptionResult.OnReceive += (res) => { Console.WriteLine(res.Data.messageAdded.content); };
-				Console.ReadKey();
+				var subscription = graphQLHttpClient.CreateSubscriptionStream(new GraphQLRequest(@"subscription { messageAdded{content}}"));
+				subscription.Subscribe(res => Console.WriteLine(res.Data.messageAdded.content));
 			}
 		}
 

src/GraphQL.Client/GraphQL.Client.csproj

@@ -15,7 +15,7 @@
 	<ItemGroup>
 		<PackageReference Include="System.Net.WebSockets.Client" Version="4.3.2" />
 		<PackageReference Include="System.Net.WebSockets.Client.Managed" Version="1.0.22" />
-		<PackageReference Include="System.Reactive" Version="4.1.2" />
+		<PackageReference Include="System.Reactive" Version="4.1.3" />
 	</ItemGroup>
 
 	<ItemGroup>

src/GraphQL.Client/Http/GraphQLHttpClient.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Concurrent;
-using System.Collections.Generic;
 using System.Net.Http;
 using System.Net.Http.Headers;
 using System.Net.WebSockets;
@@ -41,14 +40,14 @@ public GraphQLHttpClientOptions Options {
 		}
 
 		/// <inheritdoc />
-		[Obsolete("EXPERIMENTAL")]
 		public IObservable<Exception> WebSocketReceiveErrors => graphQlHttpWebSocket.ReceiveErrors;
 
 		#endregion
 
 		internal readonly GraphQLHttpHandler graphQLHttpHandler;
 		internal readonly GraphQLHttpWebSocket graphQlHttpWebSocket;
 		private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
+		private readonly ConcurrentDictionary<GraphQLRequest, IObservable<GraphQLResponse>> _subscriptionStreams = new ConcurrentDictionary<GraphQLRequest, IObservable<GraphQLResponse>>();
 
 		/// <summary>
 		/// Initializes a new instance
@@ -119,7 +118,7 @@ public Task<GraphQLResponse> SendQueryAsync(string query, CancellationToken canc
 		public Task<GraphQLResponse> SendQueryAsync(GraphQLRequest request, CancellationToken cancellationToken = default)
 		{
 			return Options.UseWebSocketForQueriesAndMutations
-				? this.graphQlHttpWebSocket.Request(request, cancellationToken)
+				? this.graphQlHttpWebSocket.SendRequest(request, cancellationToken)
 				: this.graphQLHttpHandler.PostAsync(request, cancellationToken);
 		}
 
@@ -129,55 +128,32 @@ public Task<GraphQLResponse> SendMutationAsync(string query, CancellationToken c
 		public Task<GraphQLResponse> SendMutationAsync(GraphQLRequest request, CancellationToken cancellationToken = default)
 		{
 			return Options.UseWebSocketForQueriesAndMutations
-				? this.graphQlHttpWebSocket.Request(request, cancellationToken)
+				? this.graphQlHttpWebSocket.SendRequest(request, cancellationToken)
 				: this.graphQLHttpHandler.PostAsync(request, cancellationToken);
 		}
 
-		[Obsolete("EXPERIMENTAL API")]
-		public Task<IGraphQLSubscriptionResult> SendSubscribeAsync(string query, CancellationToken cancellationToken = default) =>
-			this.SendSubscribeAsync(new GraphQLRequest(query), cancellationToken);
-
-		[Obsolete("EXPERIMENTAL API")]
-		public Task<IGraphQLSubscriptionResult> SendSubscribeAsync(GraphQLRequest request, CancellationToken cancellationToken = default)
-		{
-			GraphQLHttpSubscriptionResult graphQLSubscriptionResult = _createSubscription(request, cancellationToken);
-			return Task.FromResult<IGraphQLSubscriptionResult>(graphQLSubscriptionResult);
-		}
-
-		private GraphQLHttpSubscriptionResult _createSubscription(GraphQLRequest request, CancellationToken cancellationToken)
-		{
-			if (request == null) { throw new ArgumentNullException(nameof(request)); }
-			if (request.Query == null) { throw new ArgumentNullException(nameof(request.Query)); }
-
-			var graphQLSubscriptionResult = new GraphQLHttpSubscriptionResult(_getWebSocketUri(), request);
-			graphQLSubscriptionResult.StartAsync(cancellationToken);
-			return graphQLSubscriptionResult;
-		}
-
 		private Uri _getWebSocketUri()
 		{
 			var webSocketSchema = this.EndPoint.Scheme == "https" ? "wss" : "ws";
 			return new Uri($"{webSocketSchema}://{this.EndPoint.Host}:{this.EndPoint.Port}{this.EndPoint.AbsolutePath}");
 		}
 
 		/// <inheritdoc />
-		[Obsolete("EXPERIMENTAL API")]
 		public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest request)
 		{
 			if (_disposed)
 				throw new ObjectDisposedException(nameof(GraphQLHttpClient));
 
-			if (subscriptionStreams.ContainsKey(request))
-				return subscriptionStreams[request];
+			if (_subscriptionStreams.ContainsKey(request))
+				return _subscriptionStreams[request];
 
-			var observable = graphQlHttpWebSocket.CreateSubscriptionStream(request, Options, cancellationToken: _cancellationTokenSource.Token);
+			var observable = graphQlHttpWebSocket.CreateSubscriptionStream(request, cancellationToken: _cancellationTokenSource.Token);
 
-			subscriptionStreams.TryAdd(request, observable);
+			_subscriptionStreams.TryAdd(request, observable);
 			return observable;
 		}
 
 		/// <inheritdoc />
-		[Obsolete("EXPERIMENTAL API")]
 		public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest request, Action<WebSocketException> webSocketExceptionHandler)
 		{
 			if (_disposed)
@@ -193,22 +169,19 @@ public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest requ
 		}
 
 		/// <inheritdoc />
-		[Obsolete("EXPERIMENTAL API")]
 		public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest request, Action<Exception> exceptionHandler)
 		{
 			if (_disposed)
 				throw new ObjectDisposedException(nameof(GraphQLHttpClient));
 
-			if(subscriptionStreams.ContainsKey(request))
-				return subscriptionStreams[request];
+			if(_subscriptionStreams.ContainsKey(request))
+				return _subscriptionStreams[request];
 
-			var observable = graphQlHttpWebSocket.CreateSubscriptionStream(request, Options, exceptionHandler, _cancellationTokenSource.Token);
-			subscriptionStreams.TryAdd(request, observable);
+			var observable = graphQlHttpWebSocket.CreateSubscriptionStream(request, exceptionHandler, _cancellationTokenSource.Token);
+			_subscriptionStreams.TryAdd(request, observable);
 			return observable;
 		}
-
-		private ConcurrentDictionary<GraphQLRequest, IObservable<GraphQLResponse>> subscriptionStreams = new ConcurrentDictionary<GraphQLRequest, IObservable<GraphQLResponse>>();
-
+
 		/// <summary>
 		/// Releases unmanaged resources
 		/// </summary>
@@ -224,7 +197,7 @@ public void Dispose()
 		}
 
 		private bool _disposed = false;
-		private object _disposeLocker = new object();
+		private readonly object _disposeLocker = new object();
 
 		private void _dispose()
 		{