diff --git a/GraphQL.Client.sln b/GraphQL.Client.sln
index 722a7d85..aa9d13e0 100644
--- a/GraphQL.Client.sln
+++ b/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}
diff --git a/SubscriptionIntegrationTest.ConsoleClient/Program.cs b/SubscriptionIntegrationTest.ConsoleClient/Program.cs
new file mode 100644
index 00000000..a891236a
--- /dev/null
+++ b/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 ...");
+				}
+			}
+		}
+	}
+}
diff --git a/SubscriptionIntegrationTest.ConsoleClient/SubscriptionIntegrationTest.ConsoleClient.csproj b/SubscriptionIntegrationTest.ConsoleClient/SubscriptionIntegrationTest.ConsoleClient.csproj
new file mode 100644
index 00000000..21901c2f
--- /dev/null
+++ b/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>
diff --git a/root.props b/root.props
index e2fc2d0c..d00301d1 100644
--- a/root.props
+++ b/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>
 
diff --git a/src/GraphQL.Client/GraphQL.Client.csproj b/src/GraphQL.Client/GraphQL.Client.csproj
index 200758df..97a78f0e 100644
--- a/src/GraphQL.Client/GraphQL.Client.csproj
+++ b/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" />
diff --git a/src/GraphQL.Client/Http/GraphQLHttpClient.cs b/src/GraphQL.Client/Http/GraphQLHttpClient.cs
index fe8f4710..90209d40 100644
--- a/src/GraphQL.Client/Http/GraphQLHttpClient.cs
+++ b/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();
 		}
diff --git a/src/GraphQL.Client/Http/GraphQLHttpClientOptions.cs b/src/GraphQL.Client/Http/GraphQLHttpClientOptions.cs
index 9ff6b091..38008421 100644
--- a/src/GraphQL.Client/Http/GraphQLHttpClientOptions.cs
+++ b/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());
+		};
 	}
 
 }
diff --git a/src/GraphQL.Client/Http/GraphQLHttpObservableSubscription.cs b/src/GraphQL.Client/Http/GraphQLHttpObservableSubscription.cs
deleted file mode 100644
index d3f71990..00000000
--- a/src/GraphQL.Client/Http/GraphQLHttpObservableSubscription.cs
+++ /dev/null
@@ -1,268 +0,0 @@
-using System;
-using System.Diagnostics;
-using System.Net.WebSockets;
-using System.Reactive.Linq;
-using System.Reactive.Threading.Tasks;
-using System.Runtime.Serialization;
-using System.Text;
-using System.Threading;
-using System.Threading.Tasks;
-using GraphQL.Common.Request;
-using GraphQL.Common.Response;
-using Newtonsoft.Json;
-using Newtonsoft.Json.Linq;
-
-namespace GraphQL.Client.Http {
-
-	/// <summary>
-	/// Represents the result of a subscription query
-	/// </summary>
-	[Obsolete("EXPERIMENTAL API")]
-	public class GraphQLHttpObservableSubscription : IDisposable {
-
-		private readonly ClientWebSocket clientWebSocket = new ClientWebSocket();
-		private readonly Uri webSocketUri;
-		private readonly GraphQLRequest graphQLRequest;
-		private readonly byte[] buffer = new byte[1024 * 1024];
-		private readonly ArraySegment<byte> arraySegment;
-		private readonly CancellationTokenSource _cancellationTokenSource;
-
-		private GraphQLHttpObservableSubscription(Uri webSocketUri, GraphQLRequest graphQLRequest, CancellationToken cancellationToken = default) {
-			_cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-			_cancellationTokenSource.Token.Register(Dispose);
-			this.webSocketUri = webSocketUri;
-			this.graphQLRequest = graphQLRequest;
-			this.clientWebSocket.Options.AddSubProtocol("graphql-ws");
-			arraySegment = new ArraySegment<byte>(buffer);
-		}
-
-		public async Task ConnectAsync(CancellationToken token)
-		{
-			Debug.WriteLine($"opening websocket on subscription {this.GetHashCode()}");
-			await clientWebSocket.ConnectAsync(webSocketUri, token).ConfigureAwait(false);
-			Debug.WriteLine($"connection established on subscription {this.GetHashCode()}");
-		}
-
-		public async Task<GraphQLResponse> ReceiveResultAsync()
-		{
-			var webSocketReceiveResult = await clientWebSocket.ReceiveAsync(arraySegment, _cancellationTokenSource.Token).ConfigureAwait(false);
-			var stringResult = Encoding.UTF8.GetString(arraySegment.Array, 0, webSocketReceiveResult.Count);
-			var webSocketResponse = JsonConvert.DeserializeObject<GraphQLSubscriptionResponse>(stringResult);
-			switch (webSocketResponse.Type)
-			{
-				case GQLWebSocketMessageType.GQL_COMPLETE:
-					Debug.WriteLine($"received 'complete' message on subscription {this.GetHashCode()}");
-					Dispose();
-					break;
-				case GQLWebSocketMessageType.GQL_ERROR:
-					Debug.WriteLine($"received 'error' message on subscription {this.GetHashCode()}");
-					throw new GQLSubscriptionException(webSocketResponse.Payload);
-				default:
-					Debug.WriteLine($"received payload on subscription {this.GetHashCode()}");
-					break;
-			}
-
-			return ((JObject)webSocketResponse?.Payload).ToObject<GraphQLResponse>();
-		}
-
-		public async Task CloseAsync(CancellationToken cancellationToken = default)
-		{
-			Debug.WriteLine($"closing websocket on subscription {this.GetHashCode()}");
-			if (this.clientWebSocket.State == WebSocketState.Open) {
-				await SendCloseMessageAsync(cancellationToken).ConfigureAwait(false);
-			}
-			await this.clientWebSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "", cancellationToken).ConfigureAwait(false);
-		}
-
-		private Task SendInitialMessageAsync(CancellationToken cancellationToken = default)
-		{
-			Debug.WriteLine($"sending initial message on subscription {this.GetHashCode()}");
-			var webSocketRequest = new GraphQLSubscriptionRequest
-			{
-				Id = "1",
-				Type = GQLWebSocketMessageType.GQL_START,
-				Payload = this.graphQLRequest
-			};
-			return this.SendGraphQLSubscriptionRequest(webSocketRequest, cancellationToken);
-		}
-
-		private Task SendCloseMessageAsync(CancellationToken cancellationToken = default)
-		{
-			Debug.WriteLine($"sending close message on subscription {this.GetHashCode()}");
-			var webSocketRequest = new GraphQLSubscriptionRequest
-			{
-				Id = "1",
-				Type = GQLWebSocketMessageType.GQL_STOP,
-				Payload = this.graphQLRequest
-			};
-			return this.SendGraphQLSubscriptionRequest(webSocketRequest, cancellationToken);
-		}
-
-		private Task SendGraphQLSubscriptionRequest(GraphQLSubscriptionRequest graphQLSubscriptionRequest, CancellationToken cancellationToken = default)
-		{
-			var webSocketRequestString = JsonConvert.SerializeObject(graphQLSubscriptionRequest);
-			var arraySegmentWebSocketRequest = new ArraySegment<byte>(Encoding.UTF8.GetBytes(webSocketRequestString));
-			return this.clientWebSocket.SendAsync(arraySegmentWebSocketRequest, WebSocketMessageType.Text, true, cancellationToken);
-		}
-
-		#region IDisposable
-
-		public void Dispose()
-		{
-			// Async disposal as recommended by Stephen Cleary (https://blog.stephencleary.com/2013/03/async-oop-6-disposal.html)
-			if(Disposed == null) Disposed = DisposeAsync();
-		}
-
-		public Task Disposed { get; private set; }
-
-		private async Task DisposeAsync()
-		{
-			Debug.WriteLine($"disposing subscription {this.GetHashCode()}...");
-			if(!_cancellationTokenSource.IsCancellationRequested)
-				_cancellationTokenSource.Cancel();
-			await CloseAsync().ConfigureAwait(false);
-			clientWebSocket?.Dispose();
-			_cancellationTokenSource.Dispose();
-			Debug.WriteLine($"subscription {this.GetHashCode()} disposed");
-		}
-
-		#endregion
-
-		#region Static Factories
-
-		public static IObservable<GraphQLResponse> GetSubscriptionStream(Uri webSocketUri, GraphQLRequest graphQLRequest, CancellationToken cancellationToken = default)
-		{
-			return Observable.Using(
-				token => CreateSubscription(webSocketUri, graphQLRequest, cancellationToken),
-				InitializeSubscription
-				).Publish().RefCount();
-		}
-
-
-		private static Task<GraphQLHttpObservableSubscription> CreateSubscription(Uri webSocketUri, GraphQLRequest graphQLRequest, CancellationToken cancellationToken = default)
-		{
-			var subscription = new GraphQLHttpObservableSubscription(webSocketUri, graphQLRequest, cancellationToken);
-			return Task.FromResult(subscription);
-		}
-
-		private static async Task<IObservable<GraphQLResponse>> InitializeSubscription(GraphQLHttpObservableSubscription observableSubscription, CancellationToken cancelToken)
-		{
-			await observableSubscription.ConnectAsync(cancelToken).ConfigureAwait(false);
-			await observableSubscription.SendInitialMessageAsync(cancelToken).ConfigureAwait(false);
-			return Observable.Defer(() => observableSubscription.ReceiveResultAsync().ToObservable()).Repeat();
-		}
-
-		#endregion
-
-		[Serializable]
-		public class GQLSubscriptionException : Exception
-		{
-			//
-			// For guidelines regarding the creation of new exception types, see
-			//    http://msdn.microsoft.com/library/default.asp?url=/library/en-us/cpgenref/html/cpconerrorraisinghandlingguidelines.asp
-			// and
-			//    http://msdn.microsoft.com/library/default.asp?url=/library/en-us/dncscol/html/csharp07192001.asp
-			//
-
-			public GQLSubscriptionException()
-			{
-			}
-
-			public GQLSubscriptionException(object error) : base(error.ToString())
-			{
-			}
-
-			protected GQLSubscriptionException(
-				SerializationInfo info,
-				StreamingContext context) : base(info, context)
-			{
-			}
-		}
-
-		public static class GQLWebSocketMessageType
-		{
-
-			/// <summary>
-			///     Client sends this message after plain websocket connection to start the communication with the server
-			///     The server will response only with GQL_CONNECTION_ACK + GQL_CONNECTION_KEEP_ALIVE(if used) or GQL_CONNECTION_ERROR
-			///     to this message.
-			///     payload: Object : optional parameters that the client specifies in connectionParams
-			/// </summary>
-			public const string GQL_CONNECTION_INIT = "connection_init";
-
-			/// <summary>
-			///     The server may responses with this message to the GQL_CONNECTION_INIT from client, indicates the server accepted
-			///     the connection.
-			/// </summary>
-			public const string GQL_CONNECTION_ACK = "connection_ack"; // Server -> Client
-
-			/// <summary>
-			///     The server may responses with this message to the GQL_CONNECTION_INIT from client, indicates the server rejected
-			///     the connection.
-			///     It server also respond with this message in case of a parsing errors of the message (which does not disconnect the
-			///     client, just ignore the message).
-			///     payload: Object: the server side error
-			/// </summary>
-			public const string GQL_CONNECTION_ERROR = "connection_error"; // Server -> Client
-
-			/// <summary>
-			///     Server message that should be sent right after each GQL_CONNECTION_ACK processed and then periodically to keep the
-			///     client connection alive.
-			///     The client starts to consider the keep alive message only upon the first received keep alive message from the
-			///     server.
-			///     <remarks>
-			///         NOTE: This one here don't follow the standard due to connection optimization
-			///     </remarks>
-			/// </summary>
-			public const string GQL_CONNECTION_KEEP_ALIVE = "ka"; // Server -> Client
-
-			/// <summary>
-			///     Client sends this message in order to stop a running GraphQL operation execution (for example: unsubscribe)
-			///     id: string : operation id
-			/// </summary>
-			public const string GQL_CONNECTION_TERMINATE = "connection_terminate"; // Client -> Server
-
-			/// <summary>
-			///     Client sends this message to execute GraphQL operation
-			///     id: string : The id of the GraphQL operation to start
-			///     payload: Object:
-			///     query: string : GraphQL operation as string or parsed GraphQL document node
-			///     variables?: Object : Object with GraphQL variables
-			///     operationName?: string : GraphQL operation name
-			/// </summary>
-			public const string GQL_START = "start";
-
-			/// <summary>
-			///     The server sends this message to transfer the GraphQL execution result from the server to the client, this message
-			///     is a response for GQL_START message.
-			///     For each GraphQL operation send with GQL_START, the server will respond with at least one GQL_DATA message.
-			///     id: string : ID of the operation that was successfully set up
-			///     payload: Object :
-			///     data: any: Execution result
-			///     errors?: Error[] : Array of resolvers errors
-			/// </summary>
-			public const string GQL_DATA = "data"; // Server -> Client
-
-			/// <summary>
-			///     Server sends this message upon a failing operation, before the GraphQL execution, usually due to GraphQL validation
-			///     errors (resolver errors are part of GQL_DATA message, and will be added as errors array)
-			///     payload: Error : payload with the error attributed to the operation failing on the server
-			///     id: string : operation ID of the operation that failed on the server
-			/// </summary>
-			public const string GQL_ERROR = "error"; // Server -> Client
-
-			/// <summary>
-			///     Server sends this message to indicate that a GraphQL operation is done, and no more data will arrive for the
-			///     specific operation.
-			///     id: string : operation ID of the operation that completed
-			/// </summary>
-			public const string GQL_COMPLETE = "complete"; // Server -> Client
-
-			/// <summary>
-			///     Client sends this message in order to stop a running GraphQL operation execution (for example: unsubscribe)
-			///     id: string : operation id
-			/// </summary>
-			public const string GQL_STOP = "stop"; // Client -> Server
-		}
-	}
-}
diff --git a/src/GraphQL.Client/Http/GraphQLHttpSubscriptionHelpers.cs b/src/GraphQL.Client/Http/GraphQLHttpSubscriptionHelpers.cs
new file mode 100644
index 00000000..512dc443
--- /dev/null
+++ b/src/GraphQL.Client/Http/GraphQLHttpSubscriptionHelpers.cs
@@ -0,0 +1,144 @@
+using System;
+using System.Diagnostics;
+using System.Net.WebSockets;
+using System.Reactive.Disposables;
+using System.Reactive.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using GraphQL.Common;
+using GraphQL.Common.Request;
+using GraphQL.Common.Response;
+using Newtonsoft.Json.Linq;
+
+namespace GraphQL.Client.Http
+{
+	public static class GraphQLHttpSubscriptionHelpers
+	{
+		internal static IObservable<GraphQLResponse> CreateSubscriptionStream(
+			GraphQLRequest request,
+			GraphQLHttpWebSocket graphQlHttpWebSocket,
+			GraphQLHttpClientOptions options,
+			Action<Exception> exceptionHandler = null,
+			CancellationToken cancellationToken = default)
+		{
+			return Observable.Defer(() =>
+				Observable.Create<GraphQLResponse>(async observer =>
+				{
+					var startRequest = new GraphQLWebSocketRequest
+					{
+						Id = Guid.NewGuid().ToString("N"),
+						Type = GQLWebSocketMessageType.GQL_START,
+						Payload = request
+					};
+					var closeRequest = new GraphQLWebSocketRequest
+					{
+						Id = startRequest.Id,
+						Type = GQLWebSocketMessageType.GQL_STOP
+					};
+					var observable = graphQlHttpWebSocket.ResponseStream
+						.Where(response => response.Id == startRequest.Id)
+						.SelectMany(response =>
+						{
+							switch (response.Type)
+							{
+								case GQLWebSocketMessageType.GQL_COMPLETE:
+									Debug.WriteLine($"received 'complete' message on subscription {startRequest.Id}");
+									return Observable.Empty<GraphQLResponse>();
+								case GQLWebSocketMessageType.GQL_ERROR:
+									Debug.WriteLine($"received 'error' message on subscription {startRequest.Id}");
+									return Observable.Throw<GraphQLResponse>(
+										new GraphQLSubscriptionException(response.Payload));
+								default:
+									Debug.WriteLine($"received payload on subscription {startRequest.Id}");
+									return Observable.Return(((JObject) response?.Payload)
+										?.ToObject<GraphQLResponse>());
+							}
+						});
+
+					try
+					{
+						// intialize websocket (completes immediately if socket is already open)
+						await graphQlHttpWebSocket.InitializeWebSocket().ConfigureAwait(false);
+					}
+					catch (Exception e)
+					{
+						// subscribe observer to failed observable
+						return Observable.Throw<GraphQLResponse>(e).Subscribe(observer);
+					}
+
+					var disposable = new CompositeDisposable(
+						observable.Subscribe(observer),
+						Disposable.Create(async () =>
+						{
+							// only try to send close request on open websocket
+							if (graphQlHttpWebSocket.WebSocketState != WebSocketState.Open) return;
+
+							try
+							{
+								Debug.WriteLine($"sending close message on subscription {startRequest.Id}");
+								await graphQlHttpWebSocket.SendWebSocketRequest(closeRequest).ConfigureAwait(false);
+							}
+							// do not break on disposing
+							catch (OperationCanceledException) { }
+						})
+					);
+
+					Debug.WriteLine($"sending initial message on subscription {startRequest.Id}");
+					// send subscription request
+					try
+					{
+						await graphQlHttpWebSocket.SendWebSocketRequest(startRequest).ConfigureAwait(false);
+					}
+					catch (Exception e)
+					{
+						Console.WriteLine(e);
+						throw;
+					}
+
+					return disposable;
+				}))
+				// complete sequence on OperationCanceledException, this is triggered by the cancellation token
+				.Catch<GraphQLResponse, OperationCanceledException>(exception =>
+					Observable.Empty<GraphQLResponse>())
+				// wrap results
+				.Select(response => new Tuple<GraphQLResponse, Exception>(response, null))
+				// do exception handling
+				.Catch<Tuple<GraphQLResponse, Exception>, Exception>(e =>
+				{
+					try
+					{
+						// if the external handler is not set, propagate all exceptions (default subscription behaviour without Retry())
+						if (exceptionHandler == null) throw e;
+
+						// exceptions thrown by the handler will propagate to OnError()
+						exceptionHandler?.Invoke(e);
+
+						// throw exception on the observable to be caught by Retry() or complete sequence if cancellation was requested
+						return cancellationToken.IsCancellationRequested
+							? Observable.Empty<Tuple<GraphQLResponse, Exception>>()
+							: Observable.Throw<Tuple<GraphQLResponse, Exception>>(e);
+					}
+					catch (Exception exception)
+					{
+						// wrap all other exceptions to be propagated behind retry
+						return Observable.Return(new Tuple<GraphQLResponse, Exception>(null, exception));
+					}
+				})
+				// attempt to recreate the websocket for rethrown exceptions
+				.Retry()
+				// unwrap and push results or throw wrapped exceptions
+				.SelectMany(t =>
+				{
+					// if the result contains an exception, throw it on the observable
+					if (t.Item2 != null)
+						return Observable.Throw<GraphQLResponse>(t.Item2);
+
+					return t.Item1 == null
+						? Observable.Empty<GraphQLResponse>()
+						: Observable.Return(t.Item1);
+				})
+				// transform to hot observable and auto-connect
+				.Publish().RefCount();
+		}
+	}
+}
diff --git a/src/GraphQL.Client/Http/GraphQLHttpSubscriptionResult.cs b/src/GraphQL.Client/Http/GraphQLHttpSubscriptionResult.cs
index e0c2dc6d..34ab6f0a 100644
--- a/src/GraphQL.Client/Http/GraphQLHttpSubscriptionResult.cs
+++ b/src/GraphQL.Client/Http/GraphQLHttpSubscriptionResult.cs
@@ -3,6 +3,7 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using GraphQL.Common;
 using GraphQL.Common.Request;
 using GraphQL.Common.Response;
 using Newtonsoft.Json;
@@ -38,7 +39,7 @@ public async void StartAsync(CancellationToken cancellationToken = default) {
 				while (this.clientWebSocket.State == WebSocketState.Open) {
 					var webSocketReceiveResult = await this.clientWebSocket.ReceiveAsync(arraySegment, cancellationToken);
 					var stringResult = Encoding.UTF8.GetString(arraySegment.Array, 0, webSocketReceiveResult.Count);
-					var webSocketResponse = JsonConvert.DeserializeObject<GraphQLSubscriptionResponse>(stringResult);
+					var webSocketResponse = JsonConvert.DeserializeObject<GraphQLWebSocketResponse>(stringResult);
 					if (webSocketResponse != null)
 					{
 						var response = (GraphQLResponse) webSocketResponse.Payload;
@@ -62,7 +63,7 @@ public void Dispose() {
 		}
 
 		private Task SendInitialMessageAsync(CancellationToken cancellationToken = default) {
-			var webSocketRequest = new GraphQLSubscriptionRequest {
+			var webSocketRequest = new GraphQLWebSocketRequest {
 				Id = "1",
 				Type = GQLWebSocketMessageType.GQL_START,
 				Payload = this.graphQLRequest
@@ -71,7 +72,7 @@ private Task SendInitialMessageAsync(CancellationToken cancellationToken = defau
 		}
 
 		private Task SendCloseMessageAsync(CancellationToken cancellationToken = default) {
-			var webSocketRequest = new GraphQLSubscriptionRequest {
+			var webSocketRequest = new GraphQLWebSocketRequest {
 				Id = "1",
 				Type = GQLWebSocketMessageType.GQL_STOP,
 				Payload = this.graphQLRequest
@@ -79,98 +80,11 @@ private Task SendCloseMessageAsync(CancellationToken cancellationToken = default
 			return this.SendGraphQLSubscriptionRequest(webSocketRequest);
 		}
 
-		private Task SendGraphQLSubscriptionRequest(GraphQLSubscriptionRequest graphQLSubscriptionRequest, CancellationToken cancellationToken = default) {
-			var webSocketRequestString = JsonConvert.SerializeObject(graphQLSubscriptionRequest);
+		private Task SendGraphQLSubscriptionRequest(GraphQLWebSocketRequest graphQlWebSocketRequest, CancellationToken cancellationToken = default) {
+			var webSocketRequestString = JsonConvert.SerializeObject(graphQlWebSocketRequest);
 			var arraySegmentWebSocketRequest = new ArraySegment<byte>(Encoding.UTF8.GetBytes(webSocketRequestString));
 			return this.clientWebSocket.SendAsync(arraySegmentWebSocketRequest, WebSocketMessageType.Text, true, cancellationToken);
 		}
-
-		private static class GQLWebSocketMessageType {
-
-			/// <summary>
-			///     Client sends this message after plain websocket connection to start the communication with the server
-			///     The server will response only with GQL_CONNECTION_ACK + GQL_CONNECTION_KEEP_ALIVE(if used) or GQL_CONNECTION_ERROR
-			///     to this message.
-			///     payload: Object : optional parameters that the client specifies in connectionParams
-			/// </summary>
-			public const string GQL_CONNECTION_INIT = "connection_init";
-
-			/// <summary>
-			///     The server may responses with this message to the GQL_CONNECTION_INIT from client, indicates the server accepted
-			///     the connection.
-			/// </summary>
-			public const string GQL_CONNECTION_ACK = "connection_ack"; // Server -> Client
-
-			/// <summary>
-			///     The server may responses with this message to the GQL_CONNECTION_INIT from client, indicates the server rejected
-			///     the connection.
-			///     It server also respond with this message in case of a parsing errors of the message (which does not disconnect the
-			///     client, just ignore the message).
-			///     payload: Object: the server side error
-			/// </summary>
-			public const string GQL_CONNECTION_ERROR = "connection_error"; // Server -> Client
-
-			/// <summary>
-			///     Server message that should be sent right after each GQL_CONNECTION_ACK processed and then periodically to keep the
-			///     client connection alive.
-			///     The client starts to consider the keep alive message only upon the first received keep alive message from the
-			///     server.
-			///     <remarks>
-			///         NOTE: This one here don't follow the standard due to connection optimization
-			///     </remarks>
-			/// </summary>
-			public const string GQL_CONNECTION_KEEP_ALIVE = "ka"; // Server -> Client
-
-			/// <summary>
-			///     Client sends this message in order to stop a running GraphQL operation execution (for example: unsubscribe)
-			///     id: string : operation id
-			/// </summary>
-			public const string GQL_CONNECTION_TERMINATE = "connection_terminate"; // Client -> Server
-
-			/// <summary>
-			///     Client sends this message to execute GraphQL operation
-			///     id: string : The id of the GraphQL operation to start
-			///     payload: Object:
-			///     query: string : GraphQL operation as string or parsed GraphQL document node
-			///     variables?: Object : Object with GraphQL variables
-			///     operationName?: string : GraphQL operation name
-			/// </summary>
-			public const string GQL_START = "start";
-
-			/// <summary>
-			///     The server sends this message to transfer the GraphQL execution result from the server to the client, this message
-			///     is a response for GQL_START message.
-			///     For each GraphQL operation send with GQL_START, the server will respond with at least one GQL_DATA message.
-			///     id: string : ID of the operation that was successfully set up
-			///     payload: Object :
-			///     data: any: Execution result
-			///     errors?: Error[] : Array of resolvers errors
-			/// </summary>
-			public const string GQL_DATA = "data"; // Server -> Client
-
-			/// <summary>
-			///     Server sends this message upon a failing operation, before the GraphQL execution, usually due to GraphQL validation
-			///     errors (resolver errors are part of GQL_DATA message, and will be added as errors array)
-			///     payload: Error : payload with the error attributed to the operation failing on the server
-			///     id: string : operation ID of the operation that failed on the server
-			/// </summary>
-			public const string GQL_ERROR = "error"; // Server -> Client
-
-			/// <summary>
-			///     Server sends this message to indicate that a GraphQL operation is done, and no more data will arrive for the
-			///     specific operation.
-			///     id: string : operation ID of the operation that completed
-			/// </summary>
-			public const string GQL_COMPLETE = "complete"; // Server -> Client
-
-			/// <summary>
-			///     Client sends this message in order to stop a running GraphQL operation execution (for example: unsubscribe)
-			///     id: string : operation id
-			/// </summary>
-			public const string GQL_STOP = "stop"; // Client -> Server
-
-		}
-
 	}
 
 }
diff --git a/src/GraphQL.Client/Http/GraphQLHttpWebSocket.cs b/src/GraphQL.Client/Http/GraphQLHttpWebSocket.cs
new file mode 100644
index 00000000..c5e2d349
--- /dev/null
+++ b/src/GraphQL.Client/Http/GraphQLHttpWebSocket.cs
@@ -0,0 +1,213 @@
+using System;
+using System.Diagnostics;
+using System.Net.WebSockets;
+using System.Reactive.Disposables;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using System.Reactive.Threading.Tasks;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using GraphQL.Common.Request;
+using GraphQL.Common.Response;
+using Newtonsoft.Json;
+
+namespace GraphQL.Client.Http
+{
+	internal class GraphQLHttpWebSocket: IDisposable
+	{
+		private readonly Uri webSocketUri;
+		private readonly GraphQLHttpClientOptions _options;
+		private readonly byte[] buffer = new byte[1024 * 1024];
+		private readonly ArraySegment<byte> arraySegment;
+		private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
+
+		private Subject<GraphQLWebSocketResponse> _responseSubject = new Subject<GraphQLWebSocketResponse>();
+
+		public WebSocketState WebSocketState => clientWebSocket?.State ?? WebSocketState.None;
+
+		private ClientWebSocket clientWebSocket = null;
+		private int _connectionAttempt = 0;
+
+		public GraphQLHttpWebSocket(Uri webSocketUri, GraphQLHttpClientOptions options)
+		{
+			this.webSocketUri = webSocketUri;
+			_options = options;
+			arraySegment = new ArraySegment<byte>(buffer);
+			ResponseStream = _createResponseStream();
+		}
+
+		public IObservable<GraphQLWebSocketResponse> ResponseStream { get; }
+
+
+		public async Task SendWebSocketRequest(GraphQLWebSocketRequest request)
+		{
+			await InitializeWebSocket().ConfigureAwait(false);
+			var webSocketRequestString = JsonConvert.SerializeObject(request);
+			var arraySegmentWebSocketRequest = new ArraySegment<byte>(Encoding.UTF8.GetBytes(webSocketRequestString));
+			await this.clientWebSocket.SendAsync(arraySegmentWebSocketRequest, WebSocketMessageType.Text, true, _cancellationTokenSource.Token).ConfigureAwait(false);
+		}
+
+		public Task InitializeWebSocketTask { get; private set; } = Task.CompletedTask;
+
+		#region Private Methods
+
+		private Task _backOff()
+		{
+			_connectionAttempt++;
+
+			if(_connectionAttempt == 1) return Task.CompletedTask;
+
+			var delay = _options.BackOffStrategy(_connectionAttempt - 1);
+			Debug.WriteLine($"connection attempt #{_connectionAttempt}, backing off for {delay.TotalSeconds} s");
+			return Task.Delay(delay);
+		}
+
+		public Task InitializeWebSocket()
+		{
+			// do not attempt to initialize if cancellation is requested
+			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
+			clientWebSocket?.Dispose();
+			clientWebSocket = new ClientWebSocket();
+			this.clientWebSocket.Options.AddSubProtocol("graphql-ws");
+			return InitializeWebSocketTask = _connectAsync(_cancellationTokenSource.Token);
+		}
+
+		private IObservable<GraphQLWebSocketResponse> _createResponseStream()
+		{
+			return Observable.Create<GraphQLWebSocketResponse>(_createResultStream)
+				// complete sequence on OperationCanceledException, this is triggered by the cancellation token on disposal
+				.Catch<GraphQLWebSocketResponse, OperationCanceledException>(exception =>
+					Observable.Empty<GraphQLWebSocketResponse>());
+		}
+
+		private async Task<IDisposable> _createResultStream(IObserver<GraphQLWebSocketResponse> observer, CancellationToken token)
+		{
+			var observable = await _getReceiveResultStream().ConfigureAwait(false);
+			return new CompositeDisposable
+			(
+				observable.Subscribe(observer),
+				Disposable.Create(() =>
+				{
+					Debug.WriteLine("response stream disposed");
+				})
+			);
+		}
+
+		private async Task<IObservable<GraphQLWebSocketResponse>> _getReceiveResultStream()
+		{
+			await InitializeWebSocket().ConfigureAwait(false);
+			return Observable.Defer(() => _getReceiveTask().ToObservable()).Repeat();
+		}
+
+		private async Task _connectAsync(CancellationToken token)
+		{
+			await _backOff().ConfigureAwait(false);
+			Debug.WriteLine($"opening websocket {clientWebSocket.GetHashCode()}");
+			await clientWebSocket.ConnectAsync(webSocketUri, token).ConfigureAwait(false);
+			Debug.WriteLine($"connection established on websocket {clientWebSocket.GetHashCode()}");
+			_connectionAttempt = 1;
+		}
+
+
+		private Task<GraphQLWebSocketResponse> _receiveAsyncTask = null;
+		private object _receiveTaskLocker = new object();
+		/// <summary>
+		/// wrapper method to pick up the existing request task if already running
+		/// </summary>
+		/// <returns></returns>
+		private Task<GraphQLWebSocketResponse> _getReceiveTask()
+		{
+			lock (_receiveTaskLocker)
+			{
+				if (_receiveAsyncTask == null ||
+				    _receiveAsyncTask.IsFaulted ||
+				    _receiveAsyncTask.IsCompleted)
+					_receiveAsyncTask = _receiveResultAsync();
+			}
+
+			return _receiveAsyncTask;
+		}
+
+		private async Task<GraphQLWebSocketResponse> _receiveResultAsync()
+		{
+			try
+			{
+				Debug.WriteLine("receiving websocket data ...");
+				_cancellationTokenSource.Token.ThrowIfCancellationRequested();
+				var webSocketReceiveResult =
+					await clientWebSocket.ReceiveAsync(arraySegment, CancellationToken.None).ConfigureAwait(false);
+				_cancellationTokenSource.Token.ThrowIfCancellationRequested();
+				var stringResult = Encoding.UTF8.GetString(arraySegment.Array, 0, webSocketReceiveResult.Count);
+				return JsonConvert.DeserializeObject<GraphQLWebSocketResponse>(stringResult);
+			}
+			catch (Exception e)
+			{
+				Console.WriteLine(e);
+				throw;
+			}
+			finally
+			{
+				Debug.WriteLine("websocket data received");
+			}
+		}
+
+		private async Task _closeAsync(CancellationToken cancellationToken = default)
+		{
+			if(clientWebSocket == null)
+				return;
+
+			// don't attempt to close the websocket if it is in a failed state
+			if (this.clientWebSocket.State != WebSocketState.Open &&
+			    this.clientWebSocket.State != WebSocketState.CloseReceived &&
+			    this.clientWebSocket.State != WebSocketState.CloseSent)
+			{
+				Debug.WriteLine($"websocket {clientWebSocket.GetHashCode()} state = {this.clientWebSocket.State}");
+				return;
+			}
+
+			Debug.WriteLine($"closing websocket {clientWebSocket.GetHashCode()}");
+			await this.clientWebSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, "", cancellationToken).ConfigureAwait(false);
+		}
+
+		#endregion
+
+		#region IDisposable
+
+		private Task _disposed;
+		private object _disposedLocker = new object();
+		public void Dispose()
+		{
+			// Async disposal as recommended by Stephen Cleary (https://blog.stephencleary.com/2013/03/async-oop-6-disposal.html)
+			lock (_disposedLocker)
+			{
+				if (_disposed == null) _disposed = DisposeAsync();
+			}
+		}
+
+		private async Task DisposeAsync()
+		{
+			Debug.WriteLine($"disposing websocket {clientWebSocket.GetHashCode()}...");
+			if (!_cancellationTokenSource.IsCancellationRequested)
+				_cancellationTokenSource.Cancel();
+			await _closeAsync().ConfigureAwait(false);
+			clientWebSocket?.Dispose();
+			_cancellationTokenSource.Dispose();
+			Debug.WriteLine($"websocket {clientWebSocket.GetHashCode()} disposed");
+		}
+		#endregion
+	}
+}
diff --git a/src/GraphQL.Client/Http/GraphQLSubscriptionException.cs b/src/GraphQL.Client/Http/GraphQLSubscriptionException.cs
new file mode 100644
index 00000000..7aad64f0
--- /dev/null
+++ b/src/GraphQL.Client/Http/GraphQLSubscriptionException.cs
@@ -0,0 +1,32 @@
+using System;
+using System.Collections.Generic;
+using System.Runtime.Serialization;
+using System.Text;
+
+namespace GraphQL.Client.Http
+{
+	[Serializable]
+	public class GraphQLSubscriptionException : Exception
+	{
+		//
+		// For guidelines regarding the creation of new exception types, see
+		//    http://msdn.microsoft.com/library/default.asp?url=/library/en-us/cpgenref/html/cpconerrorraisinghandlingguidelines.asp
+		// and
+		//    http://msdn.microsoft.com/library/default.asp?url=/library/en-us/dncscol/html/csharp07192001.asp
+		//
+
+		public GraphQLSubscriptionException()
+		{
+		}
+
+		public GraphQLSubscriptionException(object error) : base(error.ToString())
+		{
+		}
+
+		protected GraphQLSubscriptionException(
+			SerializationInfo info,
+			StreamingContext context) : base(info, context)
+		{
+		}
+	}
+}
diff --git a/src/GraphQL.Client/IGraphQLClient.cs b/src/GraphQL.Client/IGraphQLClient.cs
index 3343da14..60af36ab 100644
--- a/src/GraphQL.Client/IGraphQLClient.cs
+++ b/src/GraphQL.Client/IGraphQLClient.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Net.WebSockets;
 using System.Threading;
 using System.Threading.Tasks;
 using GraphQL.Common.Request;
@@ -36,8 +37,41 @@ public interface IGraphQLClient : IDisposable {
 		[Obsolete("EXPERIMENTAL")]
 		Task<IGraphQLSubscriptionResult> SendSubscribeAsync(GraphQLRequest request, CancellationToken cancellationToken = default);
 
+		/// <summary>
+		/// Creates a subscription to a GraphQL server. The connection is not established until the first actual subscription is made.<br/>
+		/// All subscriptions made to this stream share the same hot observable.<br/>
+		/// The stream must be recreated completely after an error has occured within its logic (i.e. a <see cref="WebSocketException"/>)
+		/// </summary>
+		/// <param name="request">the GraphQL request for this subscription</param>
+		/// <returns>an observable stream for the specified subscription</returns>
 		[Obsolete("EXPERIMENTAL")]
 		IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest request);
+
+		/// <summary>
+		/// Creates a subscription to a GraphQL server. The connection is not established until the first actual subscription is made.<br/>
+		/// All subscriptions made to this stream share the same hot observable.<br/>
+		/// All <see cref="WebSocketException"/>s are passed to the <paramref name="webSocketExceptionHandler"/> to be handled externally.<br/>
+		/// If the <paramref name="webSocketExceptionHandler"/> completes normally, the subscription is recreated with a new connection attempt.<br/>
+		/// Other <see cref="Exception"/>s or any exception thrown by <paramref name="webSocketExceptionHandler"/> will cause the sequence to fail.
+		/// </summary>
+		/// <param name="request">the GraphQL request for this subscription</param>
+		/// <param name="webSocketExceptionHandler">an external handler for all <see cref="WebSocketException"/>s occuring within the sequence</param>
+		/// <returns>an observable stream for the specified subscription</returns>
+		[Obsolete("EXPERIMENTAL")]
+		IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest request, Action<WebSocketException> webSocketExceptionHandler);
+
+		/// <summary>
+		/// Creates a subscription to a GraphQL server. The connection is not established until the first actual subscription is made.<br/>
+		/// All subscriptions made to this stream share the same hot observable.<br/>
+		/// All <see cref="Exception"/>s are passed to the <paramref name="exceptionHandler"/> to be handled externally.<br/>
+		/// If the <paramref name="exceptionHandler"/> completes normally, the subscription is recreated with a new connection attempt.<br/>
+		/// Any exception thrown by <paramref name="exceptionHandler"/> will cause the sequence to fail.
+		/// </summary>
+		/// <param name="request">the GraphQL request for this subscription</param>
+		/// <param name="exceptionHandler">an external handler for all <see cref="Exception"/>s occuring within the sequence</param>
+		/// <returns>an observable stream for the specified subscription</returns>
+		[Obsolete("EXPERIMENTAL")]
+		IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest request, Action<Exception> exceptionHandler);
 	}
 
 }
diff --git a/src/GraphQL.Common/GQLWebSocketMessageType.cs b/src/GraphQL.Common/GQLWebSocketMessageType.cs
new file mode 100644
index 00000000..1e2fff76
--- /dev/null
+++ b/src/GraphQL.Common/GQLWebSocketMessageType.cs
@@ -0,0 +1,93 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace GraphQL.Common
+{
+	public static class GQLWebSocketMessageType
+	{
+
+		/// <summary>
+		///     Client sends this message after plain websocket connection to start the communication with the server
+		///     The server will response only with GQL_CONNECTION_ACK + GQL_CONNECTION_KEEP_ALIVE(if used) or GQL_CONNECTION_ERROR
+		///     to this message.
+		///     payload: Object : optional parameters that the client specifies in connectionParams
+		/// </summary>
+		public const string GQL_CONNECTION_INIT = "connection_init";
+
+		/// <summary>
+		///     The server may responses with this message to the GQL_CONNECTION_INIT from client, indicates the server accepted
+		///     the connection.
+		/// </summary>
+		public const string GQL_CONNECTION_ACK = "connection_ack"; // Server -> Client
+
+		/// <summary>
+		///     The server may responses with this message to the GQL_CONNECTION_INIT from client, indicates the server rejected
+		///     the connection.
+		///     It server also respond with this message in case of a parsing errors of the message (which does not disconnect the
+		///     client, just ignore the message).
+		///     payload: Object: the server side error
+		/// </summary>
+		public const string GQL_CONNECTION_ERROR = "connection_error"; // Server -> Client
+
+		/// <summary>
+		///     Server message that should be sent right after each GQL_CONNECTION_ACK processed and then periodically to keep the
+		///     client connection alive.
+		///     The client starts to consider the keep alive message only upon the first received keep alive message from the
+		///     server.
+		///     <remarks>
+		///         NOTE: This one here don't follow the standard due to connection optimization
+		///     </remarks>
+		/// </summary>
+		public const string GQL_CONNECTION_KEEP_ALIVE = "ka"; // Server -> Client
+
+		/// <summary>
+		///     Client sends this message in order to stop a running GraphQL operation execution (for example: unsubscribe)
+		///     id: string : operation id
+		/// </summary>
+		public const string GQL_CONNECTION_TERMINATE = "connection_terminate"; // Client -> Server
+
+		/// <summary>
+		///     Client sends this message to execute GraphQL operation
+		///     id: string : The id of the GraphQL operation to start
+		///     payload: Object:
+		///     query: string : GraphQL operation as string or parsed GraphQL document node
+		///     variables?: Object : Object with GraphQL variables
+		///     operationName?: string : GraphQL operation name
+		/// </summary>
+		public const string GQL_START = "start";
+
+		/// <summary>
+		///     The server sends this message to transfer the GraphQL execution result from the server to the client, this message
+		///     is a response for GQL_START message.
+		///     For each GraphQL operation send with GQL_START, the server will respond with at least one GQL_DATA message.
+		///     id: string : ID of the operation that was successfully set up
+		///     payload: Object :
+		///     data: any: Execution result
+		///     errors?: Error[] : Array of resolvers errors
+		/// </summary>
+		public const string GQL_DATA = "data"; // Server -> Client
+
+		/// <summary>
+		///     Server sends this message upon a failing operation, before the GraphQL execution, usually due to GraphQL validation
+		///     errors (resolver errors are part of GQL_DATA message, and will be added as errors array)
+		///     payload: Error : payload with the error attributed to the operation failing on the server
+		///     id: string : operation ID of the operation that failed on the server
+		/// </summary>
+		public const string GQL_ERROR = "error"; // Server -> Client
+
+		/// <summary>
+		///     Server sends this message to indicate that a GraphQL operation is done, and no more data will arrive for the
+		///     specific operation.
+		///     id: string : operation ID of the operation that completed
+		/// </summary>
+		public const string GQL_COMPLETE = "complete"; // Server -> Client
+
+		/// <summary>
+		///     Client sends this message in order to stop a running GraphQL operation execution (for example: unsubscribe)
+		///     id: string : operation id
+		/// </summary>
+		public const string GQL_STOP = "stop"; // Client -> Server
+
+	}
+}
diff --git a/src/GraphQL.Common/Request/GraphQLSubscriptionRequest.cs b/src/GraphQL.Common/Request/GraphQLWebSocketRequest.cs
similarity index 73%
rename from src/GraphQL.Common/Request/GraphQLSubscriptionRequest.cs
rename to src/GraphQL.Common/Request/GraphQLWebSocketRequest.cs
index a468a0b3..46e14dce 100644
--- a/src/GraphQL.Common/Request/GraphQLSubscriptionRequest.cs
+++ b/src/GraphQL.Common/Request/GraphQLWebSocketRequest.cs
@@ -7,7 +7,7 @@ namespace GraphQL.Common.Request {
 	/// A Subscription Request
 	/// </summary>
 	[Obsolete("EXPERIMENTAL")]
-	public class GraphQLSubscriptionRequest : IEquatable<GraphQLSubscriptionRequest> {
+	public class GraphQLWebSocketRequest : IEquatable<GraphQLWebSocketRequest> {
 
 		/// <summary>
 		/// The Identifier of the Response
@@ -25,10 +25,10 @@ public class GraphQLSubscriptionRequest : IEquatable<GraphQLSubscriptionRequest>
 		public GraphQLRequest Payload { get; set; }
 
 		/// <inheritdoc />
-		public override bool Equals(object obj) => this.Equals(obj as GraphQLSubscriptionRequest);
+		public override bool Equals(object obj) => this.Equals(obj as GraphQLWebSocketRequest);
 
 		/// <inheritdoc />
-		public bool Equals(GraphQLSubscriptionRequest other) {
+		public bool Equals(GraphQLWebSocketRequest other) {
 			if (other == null) {
 				return false;
 			}
@@ -57,10 +57,10 @@ public override int GetHashCode() {
 		}
 
 		/// <inheritdoc />
-		public static bool operator ==(GraphQLSubscriptionRequest request1, GraphQLSubscriptionRequest request2) => EqualityComparer<GraphQLSubscriptionRequest>.Default.Equals(request1, request2);
+		public static bool operator ==(GraphQLWebSocketRequest request1, GraphQLWebSocketRequest request2) => EqualityComparer<GraphQLWebSocketRequest>.Default.Equals(request1, request2);
 
 		/// <inheritdoc />
-		public static bool operator !=(GraphQLSubscriptionRequest request1, GraphQLSubscriptionRequest request2) => !(request1 == request2);
+		public static bool operator !=(GraphQLWebSocketRequest request1, GraphQLWebSocketRequest request2) => !(request1 == request2);
 
 	}
 
diff --git a/src/GraphQL.Common/Response/GraphQLSubscriptionResponse.cs b/src/GraphQL.Common/Response/GraphQLWebSocketResponse.cs
similarity index 72%
rename from src/GraphQL.Common/Response/GraphQLSubscriptionResponse.cs
rename to src/GraphQL.Common/Response/GraphQLWebSocketResponse.cs
index 71c47c72..cd3fb2f4 100644
--- a/src/GraphQL.Common/Response/GraphQLSubscriptionResponse.cs
+++ b/src/GraphQL.Common/Response/GraphQLWebSocketResponse.cs
@@ -7,7 +7,7 @@ namespace GraphQL.Common.Response {
 	/// A Subscription Response
 	/// </summary>
 	[Obsolete("EXPERIMENTAL")]
-	public class GraphQLSubscriptionResponse : IEquatable<GraphQLSubscriptionResponse> {
+	public class GraphQLWebSocketResponse : IEquatable<GraphQLWebSocketResponse> {
 
 		/// <summary>
 		/// The Identifier of the Response
@@ -25,10 +25,10 @@ public class GraphQLSubscriptionResponse : IEquatable<GraphQLSubscriptionRespons
 		public dynamic Payload { get; set; }
 
 		/// <inheritdoc />
-		public override bool Equals(object obj) => this.Equals(obj as GraphQLSubscriptionResponse);
+		public override bool Equals(object obj) => this.Equals(obj as GraphQLWebSocketResponse);
 
 		/// <inheritdoc />
-		public bool Equals(GraphQLSubscriptionResponse other) {
+		public bool Equals(GraphQLWebSocketResponse other) {
 			if (other == null) {
 				return false;
 			}
@@ -57,10 +57,10 @@ public override int GetHashCode() {
 		}
 
 		/// <inheritdoc />
-		public static bool operator ==(GraphQLSubscriptionResponse response1, GraphQLSubscriptionResponse response2) => EqualityComparer<GraphQLSubscriptionResponse>.Default.Equals(response1, response2);
+		public static bool operator ==(GraphQLWebSocketResponse response1, GraphQLWebSocketResponse response2) => EqualityComparer<GraphQLWebSocketResponse>.Default.Equals(response1, response2);
 
 		/// <inheritdoc />
-		public static bool operator !=(GraphQLSubscriptionResponse response1, GraphQLSubscriptionResponse response2) => !(response1 == response2);
+		public static bool operator !=(GraphQLWebSocketResponse response1, GraphQLWebSocketResponse response2) => !(response1 == response2);
 
 	}
 
diff --git a/tests/GraphQL.Integration.Tests/CallbackTester.cs b/tests/GraphQL.Integration.Tests/CallbackTester.cs
new file mode 100644
index 00000000..bf214d73
--- /dev/null
+++ b/tests/GraphQL.Integration.Tests/CallbackTester.cs
@@ -0,0 +1,81 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace GraphQL.Integration.Tests
+{
+	public class CallbackTester<T>
+	{
+		private ManualResetEventSlim _callbackInvoked { get; } = new ManualResetEventSlim();
+
+		/// <summary>
+		/// The timeout for <see cref="ShouldHaveReceivedUpdate"/>. Defaults to 1 s
+		/// </summary>
+		public TimeSpan Timeout { get; set; } = TimeSpan.FromSeconds(1);
+
+		/// <summary>
+		/// Indicates that an update has been received since the last <see cref="Reset"/>
+		/// </summary>
+		public bool CallbackInvoked => _callbackInvoked.IsSet;
+		/// <summary>
+		/// The last payload which was received.
+		/// </summary>
+		public T LastPayload { get; private set; }
+
+		public void Callback(T param)
+		{
+			LastPayload = param;
+			_callbackInvoked.Set();
+		}
+
+		/// <summary>
+		/// Asserts that a new update has been pushed to the <see cref="IObservable{T}"/> within the configured <see cref="Timeout"/> since the last <see cref="Reset"/>.
+		/// If supplied, the <paramref name="assertPayload"/> action is executed on the submitted payload.
+		/// </summary>
+		/// <param name="assertPayload">action to assert the contents of the payload</param>
+		public void CallbackShouldHaveBeenInvoked(Action<T> assertPayload = null, TimeSpan? timeout = null)
+		{
+			try
+			{
+				if (!_callbackInvoked.Wait(timeout ?? Timeout))
+					Assert.True(false, $"callback not invoked within {(timeout ?? Timeout).TotalSeconds} s!");
+
+				assertPayload?.Invoke(LastPayload);
+			}
+			finally
+			{
+				Reset();
+			}
+		}
+
+		/// <summary>
+		/// Asserts that no new update has been pushed within the given <paramref name="millisecondsTimeout"/> since the last <see cref="Reset"/>
+		/// </summary>
+		/// <param name="millisecondsTimeout">the time in ms in which no new update must be pushed to the <see cref="IObservable{T}"/>. defaults to 100</param>
+		public void CallbackShouldNotHaveBeenInvoked(TimeSpan? timeout = null)
+		{
+			if(!timeout.HasValue) timeout = TimeSpan.FromMilliseconds(100);
+			try
+			{
+				if (_callbackInvoked.Wait(timeout.Value))
+					Assert.True(false, "callback was inadvertently invoked pushed!");
+			}
+			finally
+			{
+				Reset();
+			}
+		}
+
+		/// <summary>
+		/// Resets the tester class. Should be called before triggering the potential update
+		/// </summary>
+		public void Reset()
+		{
+			LastPayload = default(T);
+			_callbackInvoked.Reset();
+		}
+	}
+}
diff --git a/tests/GraphQL.Integration.Tests/GraphQLClientExtensions.cs b/tests/GraphQL.Integration.Tests/GraphQLClientExtensions.cs
index e8b4babf..c7dd0568 100644
--- a/tests/GraphQL.Integration.Tests/GraphQLClientExtensions.cs
+++ b/tests/GraphQL.Integration.Tests/GraphQLClientExtensions.cs
@@ -22,7 +22,7 @@ public static async Task<GraphQLResponse> AddMessageAsync(this GraphQLHttpClient
 				{
 					input = new
 					{
-						fromId = "me",
+						fromId = "2",
 						content = message,
 						sentAt = DateTime.Now
 					}
diff --git a/tests/GraphQL.Integration.Tests/NetworkHelpers.cs b/tests/GraphQL.Integration.Tests/NetworkHelpers.cs
new file mode 100644
index 00000000..1462767e
--- /dev/null
+++ b/tests/GraphQL.Integration.Tests/NetworkHelpers.cs
@@ -0,0 +1,17 @@
+using System.Net;
+using System.Net.Sockets;
+
+namespace GraphQL.Integration.Tests
+{
+	public static class NetworkHelpers
+	{
+		public static int GetFreeTcpPortNumber()
+		{
+			var l = new TcpListener(IPAddress.Loopback, 0);
+			l.Start();
+			var port = ((IPEndPoint)l.LocalEndpoint).Port;
+			l.Stop();
+			return port;
+		}
+	}
+}
diff --git a/tests/GraphQL.Integration.Tests/ObservableTester.cs b/tests/GraphQL.Integration.Tests/ObservableTester.cs
index e7993a36..1569176d 100644
--- a/tests/GraphQL.Integration.Tests/ObservableTester.cs
+++ b/tests/GraphQL.Integration.Tests/ObservableTester.cs
@@ -20,7 +20,7 @@ public class ObservableTester<T> : IDisposable
 		public TimeSpan Timeout { get; set; } = TimeSpan.FromSeconds(1);
 
 		/// <summary>
-		/// Indicates that an update has been received since the last <see cref="Reset"/>
+		/// Indicates that an update has been received since the last <see cref="_reset"/>
 		/// </summary>
 		public bool UpdateReceived => _updateReceived.IsSet;
 		/// <summary>
@@ -51,84 +51,85 @@ public ObservableTester(IObservable<T> observable)
 		}
 
 		/// <summary>
-		/// Asserts that a new update has been pushed to the <see cref="IObservable{T}"/> within the configured <see cref="Timeout"/> since the last <see cref="Reset"/>.
+		/// Asserts that a new update has been pushed to the <see cref="IObservable{T}"/> within the configured <see cref="Timeout"/> since the last <see cref="_reset"/>.
 		/// If supplied, the <paramref name="assertPayload"/> action is executed on the submitted payload.
 		/// </summary>
 		/// <param name="assertPayload">action to assert the contents of the payload</param>
-		public void ShouldHaveReceivedUpdate(Action<T> assertPayload = null)
+		public void ShouldHaveReceivedUpdate(Action<T> assertPayload = null, TimeSpan? timeout = null)
 		{
 			try
 			{
-				if (!_updateReceived.Wait(Timeout))
-					Assert.True(false, "no update received!");
+				if (!_updateReceived.Wait(timeout ?? Timeout))
+					Assert.True(false, $"no update received within {(timeout ?? Timeout).TotalSeconds} s!");
 
 				assertPayload?.Invoke(LastPayload);
 			}
 			finally
 			{
-				Reset();
+				_reset();
 			}
 		}
 
 		/// <summary>
-		/// Asserts that no new update has been pushed within the given <paramref name="millisecondsTimeout"/> since the last <see cref="Reset"/>
+		/// Asserts that no new update has been pushed within the given <paramref name="millisecondsTimeout"/> since the last <see cref="_reset"/>
 		/// </summary>
 		/// <param name="millisecondsTimeout">the time in ms in which no new update must be pushed to the <see cref="IObservable{T}"/>. defaults to 100</param>
-		public void ShouldNotHaveReceivedUpdate(int millisecondsTimeout = 100)
+		public void ShouldNotHaveReceivedUpdate(TimeSpan? timeout = null)
 		{
+			if(!timeout.HasValue) timeout = TimeSpan.FromMilliseconds(100);
 			try
 			{
-				if (_updateReceived.Wait(100))
+				if (_updateReceived.Wait(timeout.Value))
 					Assert.True(false, "update was inadvertently pushed!");
 			}
 			finally
 			{
-				Reset();
+				_reset();
 			}
 		}
 
 		/// <summary>
-		/// Asserts that the subscription has completed within the configured <see cref="Timeout"/> since the last <see cref="Reset"/>
+		/// Asserts that the subscription has completed within the configured <see cref="Timeout"/> since the last <see cref="_reset"/>
 		/// </summary>
-		public void ShouldHaveCompleted()
+		public void ShouldHaveCompleted(TimeSpan? timeout = null)
 		{
 			try
 			{
-				if (!_completed.Wait(Timeout))
-					Assert.True(false, "subscription did not complete!");
+				if (!_completed.Wait(timeout ?? Timeout))
+					Assert.True(false, $"subscription did not complete within {(timeout ?? Timeout).TotalSeconds} s!");
 			}
 			finally
 			{
-				Reset();
+				_reset();
 			}
 		}
 
 		/// <summary>
-		/// Asserts that the subscription has completed within the configured <see cref="Timeout"/> since the last <see cref="Reset"/>
+		/// Asserts that the subscription has completed within the configured <see cref="Timeout"/> since the last <see cref="_reset"/>
 		/// </summary>
-		public void ShouldHaveThrownError(Action<Exception> assertError = null)
+		public void ShouldHaveThrownError(Action<Exception> assertError = null, TimeSpan? timeout = null)
 		{
 			try
 			{
-				if (!_error.Wait(Timeout))
-					Assert.True(false, "subscription did not throw an error!");
+				if (!_error.Wait(timeout ?? Timeout))
+					Assert.True(false, $"subscription did not throw an error within {(timeout ?? Timeout).TotalSeconds} s!");
 
 				assertError?.Invoke(Error);
 			}
 			finally
 			{
-				Reset();
+				_reset();
 			}
 		}
 
 		/// <summary>
 		/// Resets the tester class. Should be called before triggering the potential update
 		/// </summary>
-		public void Reset()
+		private void _reset()
 		{
-			if (_completed.IsSet)
-				throw new InvalidOperationException(
-					"the subscription sequence has completed. this tester instance cannot be reused");
+			//if (_completed.IsSet)
+			//	throw new InvalidOperationException(
+			//		"the subscription sequence has completed. this tester instance cannot be reused");
 
 			LastPayload = default(T);
 			_updateReceived.Reset();
diff --git a/tests/GraphQL.Integration.Tests/SubscriptionsTest.cs b/tests/GraphQL.Integration.Tests/SubscriptionsTest.cs
index f8a66760..36804c95 100644
--- a/tests/GraphQL.Integration.Tests/SubscriptionsTest.cs
+++ b/tests/GraphQL.Integration.Tests/SubscriptionsTest.cs
@@ -1,5 +1,7 @@
 using System;
 using System.Diagnostics;
+using System.Net.WebSockets;
+using System.Threading;
 using System.Threading.Tasks;
 using GraphQL.Client.Http;
 using GraphQL.Common.Request;
@@ -36,16 +38,20 @@ public static IWebHost CreateServer(int port)
 		private readonly IWebHost _server;
 
 		public SubscriptionsTest()
-		{}
+		{
+		}
 
 		private GraphQLHttpClient GetGraphQLClient(int port)
-			=> new GraphQLHttpClient($"http://localhost:{port}/graphql");
+			=> new GraphQLHttpClient(new GraphQLHttpClientOptions
+			{
+				EndPoint = new Uri($"http://localhost:{port}/graphql"),
+			} );
 
 
 		[Fact]
 		public async void AssertTestingHarness()
 		{
-			var port = 5001;
+			var port = NetworkHelpers.GetFreeTcpPortNumber();
 			using (CreateServer(port))
 			{
 				var client = GetGraphQLClient(port);
@@ -53,29 +59,31 @@ public async void AssertTestingHarness()
 				const string message = "some random testing message";
 				var response = await client.AddMessageAsync(message).ConfigureAwait(false);
 
-				Assert.Equal(message, (string)response.Data.addMessage.content);
+				Assert.Equal(message, (string) response.Data.addMessage.content);
 			}
 		}
 
+		private const string SubscriptionQuery = @"
+			subscription {
+			  messageAdded{
+			    content
+			  }
+			}";
+
+		private readonly GraphQLRequest SubscriptionRequest = new GraphQLRequest
+		{
+			Query = SubscriptionQuery
+		};
+
 		[Fact]
 		public async void CanCreateObservableSubscription()
 		{
-			var port = 5002;
+			var port = NetworkHelpers.GetFreeTcpPortNumber();
 			using (CreateServer(port))
 			{
 				var client = GetGraphQLClient(port);
-				var graphQLRequest = new GraphQLRequest
-				{
-					Query = @"
-					subscription {
-					  messageAdded{
-					    content
-					  }
-					}"
-				};
-
 				Debug.WriteLine("creating subscription stream");
-				IObservable<GraphQLResponse> observable = client.CreateSubscriptionStream(graphQLRequest);
+				IObservable<GraphQLResponse> observable = client.CreateSubscriptionStream(SubscriptionRequest);
 
 				Debug.WriteLine("subscribing...");
 				var tester = observable.SubscribeTester();
@@ -89,26 +97,195 @@ public async void CanCreateObservableSubscription()
 					Assert.Equal(message1, (string) gqlResponse.Data.messageAdded.content.Value);
 				});
 
-				tester.Reset();
-
 				const string message2 = "lorem ipsum dolor si amet";
 				response = await client.AddMessageAsync(message2).ConfigureAwait(false);
+				Assert.Equal(message2, (string) response.Data.addMessage.content);
+				tester.ShouldHaveReceivedUpdate(gqlResponse =>
+				{
+					Assert.Equal(message2, (string) gqlResponse.Data.messageAdded.content.Value);
+				});
+
+				// disposing the client should throw a TaskCanceledException on the subscription
+				client.Dispose();
+				tester.ShouldHaveCompleted();
+			}
+		}
+
+		[Fact]
+		public async void CanReconnectWithSameObservable()
+		{
+			var port = NetworkHelpers.GetFreeTcpPortNumber();
+			using (CreateServer(port))
+			{
+				var client = GetGraphQLClient(port);
+
+				Debug.WriteLine("creating subscription stream");
+				IObservable<GraphQLResponse> observable = client.CreateSubscriptionStream(SubscriptionRequest);
+
+				Debug.WriteLine("subscribing...");
+				var tester = observable.SubscribeTester();
+
+				const string message1 = "Hello World";
+				var response = await client.AddMessageAsync(message1).ConfigureAwait(false);
+				Assert.Equal(message1, (string) response.Data.addMessage.content);
+				tester.ShouldHaveReceivedUpdate(gqlResponse =>
+				{
+					Assert.Equal(message1, (string) gqlResponse.Data.messageAdded.content.Value);
+				});
+
+				const string message2 = "How are you?";
+				response = await client.AddMessageAsync(message2).ConfigureAwait(false);
+				Assert.Equal(message2, (string) response.Data.addMessage.content);
+				tester.ShouldHaveReceivedUpdate(gqlResponse =>
+				{
+					Assert.Equal(message2, (string) gqlResponse.Data.messageAdded.content.Value);
+				});
+
+				Debug.WriteLine("disposing subscription...");
+				tester.Dispose();
+
+				Debug.WriteLine("creating new subscription...");
+				tester = observable.SubscribeTester();
+				tester.ShouldHaveReceivedUpdate(
+					gqlResponse => { Assert.Equal(message2, (string) gqlResponse.Data.messageAdded.content.Value); },
+					TimeSpan.FromSeconds(10));
+				const string message3 = "lorem ipsum dolor si amet";
+				response = await client.AddMessageAsync(message3).ConfigureAwait(false);
+				Assert.Equal(message3, (string) response.Data.addMessage.content);
+				tester.ShouldHaveReceivedUpdate(gqlResponse =>
+				{
+					Assert.Equal(message3, (string) gqlResponse.Data.messageAdded.content.Value);
+				});
+
+				// disposing the client should complete the subscription
+				client.Dispose();
+				tester.ShouldHaveCompleted();
+			}
+		}
+
+		private const string SubscriptionQuery2 = @"
+			subscription {
+			  messageAdded{
+			    content
+				from {
+				  displayName
+				}
+			  }
+			}";
+
+		private readonly GraphQLRequest SubscriptionRequest2 = new GraphQLRequest
+		{
+			Query = SubscriptionQuery2
+		};
+
+		[Fact]
+		public async void CanConnectMultipleSubscriptionsSimultaneously()
+		{
+			var port = NetworkHelpers.GetFreeTcpPortNumber();
+			var callbackTester = new CallbackTester<Exception>();
+			var callbackTester2 = new CallbackTester<Exception>();
+			using (CreateServer(port))
+			{
+				var client = GetGraphQLClient(port);
+
+				Debug.WriteLine("creating subscription stream");
+				IObservable<GraphQLResponse> subscription1 = client.CreateSubscriptionStream(SubscriptionRequest, callbackTester.Callback);
+
+				Debug.WriteLine("subscribing...");
+				var tester = subscription1.SubscribeTester();
+
+				const string message1 = "Hello World";
+				var response = await client.AddMessageAsync(message1).ConfigureAwait(false);
+				Assert.Equal(message1, (string)response.Data.addMessage.content);
+				tester.ShouldHaveReceivedUpdate(gqlResponse =>
+				{
+					Assert.Equal(message1, (string)gqlResponse.Data.messageAdded.content.Value);
+				});
+
+				IObservable<GraphQLResponse> subscription2 = client.CreateSubscriptionStream(SubscriptionRequest2, callbackTester2.Callback);
+				var tester2 = subscription2.SubscribeTester();
+				tester2.ShouldHaveReceivedUpdate(gqlResponse =>
+				{
+					Assert.Equal(message1, (string)gqlResponse.Data.messageAdded.content.Value);
+					Assert.Equal("tester", (string)gqlResponse.Data.messageAdded.from.displayName.Value);
+				}, TimeSpan.FromSeconds(10));
+
+				const string message2 = "How are you?";
+				response = await client.AddMessageAsync(message2).ConfigureAwait(false);
 				Assert.Equal(message2, (string)response.Data.addMessage.content);
 				tester.ShouldHaveReceivedUpdate(gqlResponse =>
 				{
 					Assert.Equal(message2, (string)gqlResponse.Data.messageAdded.content.Value);
 				});
+				tester2.ShouldHaveReceivedUpdate(gqlResponse =>
+				{
+					Assert.Equal(message2, (string)gqlResponse.Data.messageAdded.content.Value);
+					Assert.Equal("tester", (string)gqlResponse.Data.messageAdded.from.displayName.Value);
+				});
 
-				tester.Reset();
+				Debug.WriteLine("disposing subscription...");
+				tester.Dispose();
 
-				// disposing the client should throw a TaskCanceledException on the subscription
-				client.Dispose();
-				tester.ShouldHaveThrownError(exception =>
+				const string message3 = "lorem ipsum dolor si amet";
+				response = await client.AddMessageAsync(message3).ConfigureAwait(false);
+				Assert.Equal(message3, (string)response.Data.addMessage.content);
+				tester2.ShouldHaveReceivedUpdate(gqlResponse =>
 				{
-					Assert.True(exception is TaskCanceledException,
-							$"exception is of unexpected type {exception.GetType().Name}");
+					Assert.Equal(message3, (string)gqlResponse.Data.messageAdded.content.Value);
+					Assert.Equal("tester", (string)gqlResponse.Data.messageAdded.from.displayName.Value);
 				});
+
+				// disposing the client should complete the subscription
+				client.Dispose();
+				tester2.ShouldHaveCompleted();
+			}
+		}
+
+		[Fact]
+		public async void CanHandleConnectionTimeout()
+		{
+			var port = NetworkHelpers.GetFreeTcpPortNumber();
+			var server = CreateServer(port);
+			var callbackTester = new CallbackTester<Exception>();
+
+			var client = GetGraphQLClient(port);
+			Debug.WriteLine("creating subscription stream");
+			IObservable<GraphQLResponse> observable = client.CreateSubscriptionStream(SubscriptionRequest, callbackTester.Callback);
+
+			Debug.WriteLine("subscribing...");
+			var tester = observable.SubscribeTester();
+			const string message1 = "Hello World";
+
+			var response = await client.AddMessageAsync(message1).ConfigureAwait(false);
+			Assert.Equal(message1, (string)response.Data.addMessage.content);
+
+			tester.ShouldHaveReceivedUpdate(gqlResponse =>
+			{
+				Assert.Equal(message1, (string)gqlResponse.Data.messageAdded.content.Value);
+			});
+
+			Debug.WriteLine("stopping web host...");
+			await server.StopAsync(CancellationToken.None).ConfigureAwait(false);
+			Debug.WriteLine("web host stopped...");
+
+			callbackTester.CallbackShouldHaveBeenInvoked(exception =>
+			{
+				Assert.IsType<WebSocketException>(exception);
+			}, TimeSpan.FromSeconds(10));
+
+			try
+			{
+				server.Start();
 			}
+			catch (Exception e)
+			{
+				Console.WriteLine(e);
+			}
+
+			// disposing the client should complete the subscription
+			client.Dispose();
+			tester.ShouldHaveCompleted(TimeSpan.FromSeconds(5));
+			server.Dispose();
 		}
 	}
 }