Use single websocket for all subscriptions

GraphQL.Client.sln

@@ -1,4 +1,4 @@
-
+
 Microsoft Visual Studio Solution File, Format Version 12.00
 # Visual Studio 15
 VisualStudioVersion = 15.0.27130.2010
@@ -49,6 +49,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "GraphQL.Integration.Tests",
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "IntegrationTestServer", "tests\IntegrationTestServer\IntegrationTestServer.csproj", "{618653E5-41C2-4F17-BE4F-F904267500D4}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "SubscriptionIntegrationTest.ConsoleClient", "SubscriptionIntegrationTest.ConsoleClient\SubscriptionIntegrationTest.ConsoleClient.csproj", "{D588BC10-6B17-477F-9546-F8D1CE02EACB}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -83,6 +85,10 @@ Global
 		{618653E5-41C2-4F17-BE4F-F904267500D4}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{618653E5-41C2-4F17-BE4F-F904267500D4}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{618653E5-41C2-4F17-BE4F-F904267500D4}.Release|Any CPU.Build.0 = Release|Any CPU
+		{D588BC10-6B17-477F-9546-F8D1CE02EACB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{D588BC10-6B17-477F-9546-F8D1CE02EACB}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{D588BC10-6B17-477F-9546-F8D1CE02EACB}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{D588BC10-6B17-477F-9546-F8D1CE02EACB}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -97,6 +103,7 @@ Global
 		{B21E97C3-F328-473F-A054-A4BF272B63F0} = {9413EC62-CDDE-4E77-9784-E1136EA5D1EE}
 		{86BC3878-6549-4EF1-9672-B7C15A3FDF46} = {0B0EDB0F-FF67-4B78-A8DB-B5C23E1FEE8C}
 		{618653E5-41C2-4F17-BE4F-F904267500D4} = {0B0EDB0F-FF67-4B78-A8DB-B5C23E1FEE8C}
+		{D588BC10-6B17-477F-9546-F8D1CE02EACB} = {0B0EDB0F-FF67-4B78-A8DB-B5C23E1FEE8C}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {387AC1AC-F90C-4EF8-955A-04D495C75AF4}

SubscriptionIntegrationTest.ConsoleClient/Program.cs

@@ -0,0 +1,44 @@
+using System;
+using GraphQL.Client.Http;
+using GraphQL.Common.Request;
+
+namespace SubsccriptionIntegrationTest.ConsoleClient
+{
+	class Program
+	{
+		static void Main(string[] args)
+		{
+			Console.WriteLine("configuring client ...");
+			using (var client = new GraphQLHttpClient("http://localhost:5000/graphql/"))
+			{
+#pragma warning disable 618
+				var stream = client.CreateSubscriptionStream(new GraphQLRequest
+					{
+						Query = @"
+							subscription {
+							  messageAdded{
+							    content
+								from {
+								  displayName
+								}
+							  }
+							}"
+					},
+					e => Console.WriteLine($"WebSocketException: {e.Message} (WebSocketError {e.WebSocketErrorCode}, ErrorCode {e.ErrorCode}, NativeErrorCode {e.NativeErrorCode}"));
+#pragma warning restore 618
+
+				Console.WriteLine("subscribing to message stream ...");
+				using (var subscription = stream.Subscribe(
+					response => Console.WriteLine($"new message from \"{response.Data.messageAdded.from.displayName.Value}\": {response.Data.messageAdded.content.Value}"),
+					exception => Console.WriteLine($"message subscription stream failed: {exception}"),
+					() => Console.WriteLine($"message subscription stream completed")))
+				{
+					Console.WriteLine("client setup complete");
+					Console.WriteLine("press any key to exit");
+					Console.Read();
+					Console.WriteLine("shutting down ...");
+				}
+			}
+		}
+	}
+}

SubscriptionIntegrationTest.ConsoleClient/SubscriptionIntegrationTest.ConsoleClient.csproj

@@ -0,0 +1,13 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFramework>netcoreapp2.1</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\src\GraphQL.Client\GraphQL.Client.csproj" />
+    <ProjectReference Include="..\src\GraphQL.Common\GraphQL.Common.csproj" />
+  </ItemGroup>
+
+</Project>

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.3</Version>
+		<Version>2.0.0-alpha.subscriptions-api.10</Version>
 		<WarningLevel>4</WarningLevel>
 	</PropertyGroup>
 

src/GraphQL.Client/GraphQL.Client.csproj

@@ -17,6 +17,12 @@
 		<PackageReference Include="System.Reactive" Version="4.1.2" />
 	</ItemGroup>
 
+	<ItemGroup>
+	  <Reference Include="System.Reactive">
+	    <HintPath>C:\Users\arose\.nuget\packages\system.reactive\4.1.2\lib\netstandard2.0\System.Reactive.dll</HintPath>
+	  </Reference>
+	</ItemGroup>
+
 	<!-- Allow method ReadAsAsync<T>
 	<ItemGroup>
 		<PackageReference Include="Microsoft.AspNet.WebApi.Client" Version="5.2.4" />

src/GraphQL.Client/Http/GraphQLHttpClient.cs

@@ -1,7 +1,7 @@
 using System;
 using System.Net.Http;
 using System.Net.Http.Headers;
-using System.Reactive.Linq;
+using System.Net.WebSockets;
 using System.Threading;
 using System.Threading.Tasks;
 using GraphQL.Client.Http.Internal;
@@ -41,6 +41,7 @@ public GraphQLHttpClientOptions Options {
 		#endregion
 
 		internal readonly GraphQLHttpHandler graphQLHttpHandler;
+		internal readonly GraphQLHttpWebSocket graphQlHttpWebSocket;
 		private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
 
 		/// <summary>
@@ -76,6 +77,7 @@ public GraphQLHttpClient(Uri endPoint, GraphQLHttpClientOptions options) {
 
 			options.EndPoint = endPoint ?? throw new ArgumentNullException(nameof(endPoint));
 			this.graphQLHttpHandler = new GraphQLHttpHandler(options);
+			this.graphQlHttpWebSocket = new GraphQLHttpWebSocket(_getWebSocketUri(), options);
 		}
 
 		/// <summary>
@@ -90,6 +92,7 @@ public GraphQLHttpClient(GraphQLHttpClientOptions options) {
 			if (options.MediaType == null) { throw new ArgumentNullException(nameof(options.MediaType)); }
 
 			this.graphQLHttpHandler = new GraphQLHttpHandler(options);
+			this.graphQlHttpWebSocket = new GraphQLHttpWebSocket(_getWebSocketUri(), options);
 		}
 
 		internal GraphQLHttpClient(GraphQLHttpClientOptions options, HttpClient httpClient) {
@@ -100,6 +103,7 @@ internal GraphQLHttpClient(GraphQLHttpClientOptions options, HttpClient httpClie
 			if (options.MediaType == null) { throw new ArgumentNullException(nameof(options.MediaType)); }
 
 			this.graphQLHttpHandler = new GraphQLHttpHandler(options, httpClient);
+			this.graphQlHttpWebSocket = new GraphQLHttpWebSocket(_getWebSocketUri(), options);
 		}
 
 		public Task<GraphQLResponse> SendQueryAsync(string query, CancellationToken cancellationToken = default) =>
@@ -141,10 +145,35 @@ private Uri _getWebSocketUri()
 			return new Uri($"{webSocketSchema}://{this.EndPoint.Host}:{this.EndPoint.Port}{this.EndPoint.AbsolutePath}");
 		}
 
+		/// <inheritdoc />
 		[Obsolete("EXPERIMENTAL API")]
 		public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest request)
 		{
-			var observable = GraphQLHttpObservableSubscription.GetSubscriptionStream(_getWebSocketUri(), request, _cancellationTokenSource.Token);
+			if (_disposed)
+				throw new ObjectDisposedException(nameof(GraphQLHttpClient));
+
+			return GraphQLHttpSubscriptionHelpers.CreateSubscriptionStream(request, graphQlHttpWebSocket,
+				Options, cancellationToken: _cancellationTokenSource.Token);
+		}
+
+		/// <inheritdoc />
+		[Obsolete("EXPERIMENTAL API")]
+		public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest request, Action<WebSocketException> webSocketExceptionHandler)
+		{
+			return CreateSubscriptionStream(request, e =>
+			{
+				if (e is WebSocketException webSocketException)
+					webSocketExceptionHandler(webSocketException);
+				else
+					throw e;
+			});
+		}
+
+		/// <inheritdoc />
+		[Obsolete("EXPERIMENTAL API")]
+		public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest request, Action<Exception> exceptionHandler)
+		{
+			var observable = GraphQLHttpSubscriptionHelpers.CreateSubscriptionStream(request, graphQlHttpWebSocket, Options, exceptionHandler, _cancellationTokenSource.Token);
 			return observable;
 		}
 
@@ -153,7 +182,23 @@ public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest requ
 		/// </summary>
 		public void Dispose()
 		{
+			lock (_disposeLocker)
+			{
+				if (!_disposed)
+				{
+					_dispose();
+				}
+			}
+		}
+
+		private bool _disposed = false;
+		private object _disposeLocker = new object();
+
+		private void _dispose()
+		{
+			_disposed = true;
 			this.graphQLHttpHandler.Dispose();
+			this.graphQlHttpWebSocket.Dispose();
 			_cancellationTokenSource.Cancel();
 			_cancellationTokenSource.Dispose();
 		}

src/GraphQL.Client/Http/GraphQLHttpClientOptions.cs

@@ -1,7 +1,10 @@
 using System;
+using System.Collections.Generic;
 using System.Net.Http;
 using System.Net.Http.Headers;
+using System.Net.WebSockets;
 using Newtonsoft.Json;
+using Newtonsoft.Json.Converters;
 using Newtonsoft.Json.Serialization;
 
 namespace GraphQL.Client.Http {
@@ -20,7 +23,11 @@ public class GraphQLHttpClientOptions {
 		/// The <see cref="Newtonsoft.Json.JsonSerializerSettings"/> that is going to be used
 		/// </summary>
 		public JsonSerializerSettings JsonSerializerSettings { get; set; } = new JsonSerializerSettings {
-			ContractResolver = new CamelCasePropertyNamesContractResolver()
+			ContractResolver = new CamelCasePropertyNamesContractResolver(),
+			Converters = new List<JsonConverter>
+			{
+				new StringEnumConverter()
+			}
 		};
 
 		/// <summary>
@@ -33,6 +40,15 @@ public class GraphQLHttpClientOptions {
 		/// </summary>
 		public MediaTypeHeaderValue MediaType { get; set; } = MediaTypeHeaderValue.Parse("application/json; charset=utf-8"); // This should be "application/graphql" also "application/x-www-form-urlencoded" is Accepted
 
+		/// <summary>
+		/// The back-off strategy for automatic websocket/subscription reconnects. Calculates the delay before the next connection attempt is made.<br/>
+		/// default formula: min(n, 5) * 1,5 * random(0.0, 1.0)
+		/// </summary>
+		public Func<int, TimeSpan> BackOffStrategy = n =>
+		{
+			var rnd = new Random();
+			return TimeSpan.FromSeconds(Math.Min(n, 5) * 1.5 + rnd.NextDouble());
+		};
 	}
 
 }