diff --git a/GraphQL.Client.sln b/GraphQL.Client.sln
index d0bf4f94..4d28e424 100644
--- a/GraphQL.Client.sln
+++ b/GraphQL.Client.sln
@@ -1,4 +1,4 @@
-
+
 Microsoft Visual Studio Solution File, Format Version 12.00
 # Visual Studio Version 16
 VisualStudioVersion = 16.0.28407.52
@@ -49,6 +49,12 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "benchmarks", "benchmarks",
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "GraphQL.Common.Benchmark", "benchmarks\GraphQL.Common.Benchmark\GraphQL.Common.Benchmark.csproj", "{6A935C7C-7DD1-489D-9259-56C1A398D643}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "GraphQL.Integration.Tests", "tests\GraphQL.Integration.Tests\GraphQL.Integration.Tests.csproj", "{86BC3878-6549-4EF1-9672-B7C15A3FDF46}"
+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
@@ -75,6 +81,18 @@ Global
 		{B21E97C3-F328-473F-A054-A4BF272B63F0}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{B21E97C3-F328-473F-A054-A4BF272B63F0}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{B21E97C3-F328-473F-A054-A4BF272B63F0}.Release|Any CPU.Build.0 = Release|Any CPU
+		{86BC3878-6549-4EF1-9672-B7C15A3FDF46}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{86BC3878-6549-4EF1-9672-B7C15A3FDF46}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{86BC3878-6549-4EF1-9672-B7C15A3FDF46}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{86BC3878-6549-4EF1-9672-B7C15A3FDF46}.Release|Any CPU.Build.0 = Release|Any CPU
+		{618653E5-41C2-4F17-BE4F-F904267500D4}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{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
 		{6A935C7C-7DD1-489D-9259-56C1A398D643}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{6A935C7C-7DD1-489D-9259-56C1A398D643}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{6A935C7C-7DD1-489D-9259-56C1A398D643}.Release|Any CPU.ActiveCfg = Release|Any CPU
@@ -91,6 +109,9 @@ Global
 		{6326E0E2-3F48-4BAF-80D3-47AED5EB647C} = {63F75859-4698-4EDE-8B70-4ACBB8BC425A}
 		{C1406F03-650F-4633-887D-312943251919} = {63F75859-4698-4EDE-8B70-4ACBB8BC425A}
 		{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}
 		{6A935C7C-7DD1-489D-9259-56C1A398D643} = {A2E950A3-BD50-40C0-8189-57B455FFBC62}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
diff --git a/SubscriptionIntegrationTest.ConsoleClient/Program.cs b/SubscriptionIntegrationTest.ConsoleClient/Program.cs
new file mode 100644
index 00000000..95000512
--- /dev/null
+++ b/SubscriptionIntegrationTest.ConsoleClient/Program.cs
@@ -0,0 +1,135 @@
+using System;
+using System.Net.WebSockets;
+using System.Reactive.Disposables;
+using System.Reactive.Linq;
+using System.Threading.Tasks;
+using GraphQL.Client.Http;
+using GraphQL.Common.Request;
+
+namespace SubsccriptionIntegrationTest.ConsoleClient
+{
+	class Program
+	{
+		static async Task Main(string[] args)
+		{
+			Console.WriteLine("configuring client ...");
+			using (var client = new GraphQLHttpClient("http://localhost:5000/graphql/", new GraphQLHttpClientOptions{ UseWebSocketForQueriesAndMutations = true }))
+			{
+
+				Console.WriteLine("subscribing to message stream ...");
+
+				var subscriptions = new CompositeDisposable();
+
+				subscriptions.Add(client.WebSocketReceiveErrors.Subscribe(e => {
+					if(e is WebSocketException we)
+						Console.WriteLine($"WebSocketException: {we.Message} (WebSocketError {we.WebSocketErrorCode}, ErrorCode {we.ErrorCode}, NativeErrorCode {we.NativeErrorCode}");
+					else
+						Console.WriteLine($"Exception in websocket receive stream: {e.ToString()}");
+				}));
+
+				subscriptions.Add(CreateSubscription("1", client));
+				await Task.Delay(200);
+				subscriptions.Add(CreateSubscription2("2", client));
+				await Task.Delay(200);
+				subscriptions.Add(CreateSubscription("3", client));
+				await Task.Delay(200);
+				subscriptions.Add(CreateSubscription("4", client));
+				await Task.Delay(200);
+				subscriptions.Add(CreateSubscription("5", client));
+				await Task.Delay(200);
+				subscriptions.Add(CreateSubscription("6", client));
+				await Task.Delay(200);
+				subscriptions.Add(CreateSubscription("7", client));
+
+				using (subscriptions)
+				{
+					Console.WriteLine("client setup complete");
+					var quit = false;
+					do
+					{
+						Console.WriteLine("write message and press enter...");
+						var message = Console.ReadLine();
+						var graphQLRequest = new GraphQLRequest(@"
+							mutation($input: MessageInputType){
+							  addMessage(message: $input){
+								content
+							  }
+							}")
+						{
+							Variables = new
+							{
+								input = new
+								{
+									fromId = "2",
+									content = message,
+									sentAt = DateTime.Now
+								}
+							}
+						};
+						var result = await client.SendMutationAsync(graphQLRequest).ConfigureAwait(false);
+
+						if(result.Errors != null && result.Errors.Length > 0)
+						{
+							Console.WriteLine($"request returned {result.Errors.Length} errors:");
+							foreach (var item in result.Errors)
+							{
+								Console.WriteLine($"{item.Message}");
+							}
+						}
+					}
+					while(!quit);
+					Console.WriteLine("shutting down ...");
+				}
+				Console.WriteLine("subscriptions disposed ...");
+			}
+			Console.WriteLine("client disposed ...");
+		}
+
+		private static IDisposable CreateSubscription(string id, GraphQLHttpClient client)
+		{
+#pragma warning disable 618
+			var stream = client.CreateSubscriptionStream(new GraphQLRequest(@"
+					subscription {
+						messageAdded{
+						content
+							from {
+								displayName
+							}
+						}
+					}"
+				)
+			{ Variables = new { id } });
+#pragma warning restore 618
+
+			return stream.Subscribe(
+				response => Console.WriteLine($"{id}: new message from \"{response.Data.messageAdded.from.displayName.Value}\": {response.Data.messageAdded.content.Value}"),
+				exception => Console.WriteLine($"{id}: message subscription stream failed: {exception}"),
+				() => Console.WriteLine($"{id}: message subscription stream completed"));
+			
+		}
+
+
+		private static IDisposable CreateSubscription2(string id, GraphQLHttpClient client)
+		{
+#pragma warning disable 618
+			var stream = client.CreateSubscriptionStream(new GraphQLRequest(@"
+					subscription {
+						contentAdded{
+						content
+							from {
+								displayName
+							}
+						}
+					}"
+				)
+			{ Variables = new { id } });
+#pragma warning restore 618
+
+			return stream.Subscribe(
+				response => Console.WriteLine($"{id}: new content from \"{response.Data.contentAdded.from.displayName.Value}\": {response.Data.contentAdded.content.Value}"),
+				exception => Console.WriteLine($"{id}: content subscription stream failed: {exception}"),
+				() => Console.WriteLine($"{id}: content subscription stream completed"));
+
+		}
+	}
+}
diff --git a/SubscriptionIntegrationTest.ConsoleClient/SubscriptionIntegrationTest.ConsoleClient.csproj b/SubscriptionIntegrationTest.ConsoleClient/SubscriptionIntegrationTest.ConsoleClient.csproj
new file mode 100644
index 00000000..83d4884b
--- /dev/null
+++ b/SubscriptionIntegrationTest.ConsoleClient/SubscriptionIntegrationTest.ConsoleClient.csproj
@@ -0,0 +1,18 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <TargetFrameworks>netcoreapp3.0;net461</TargetFrameworks>
+	  <LangVersion>8.0</LangVersion>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="System.Reactive" Version="4.1.2" />
+  </ItemGroup>
+
+  <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 9f8c24b7..df16b57f 100644
--- a/root.props
+++ b/root.props
@@ -8,7 +8,7 @@
 		<GenerateDocumentationFile>True</GenerateDocumentationFile>
 		<LangVersion>8.0</LangVersion>
 		<NeutralLanguage>en-US</NeutralLanguage>
-		<NoWarn>CS0618;CS1591;CS1701</NoWarn>
+		<NoWarn>CS0618;CS1591;CS1701;NU5105;NU5125</NoWarn>
 		<PackageIconUrl>https://raw.githubusercontent.com/graphql-dotnet/graphql-client/master/assets/logo.64x64.png</PackageIconUrl>
 		<PackageLicenseUrl>https://raw.githubusercontent.com/graphql-dotnet/graphql-client/master/LICENSE.txt</PackageLicenseUrl>
 		<PackageProjectUrl>https://github.com/graphql-dotnet/graphql-client</PackageProjectUrl>
@@ -17,7 +17,7 @@
 		<RepositoryType>git</RepositoryType>
 		<RepositoryUrl>https://github.com/graphql-dotnet/graphql-client.git</RepositoryUrl>
 		<TreatWarningsAsErrors>True</TreatWarningsAsErrors>
-		<Version>2.0.0-alpha.4</Version>
+		<Version>2.0.0-alpha.4.subscription-api.10</Version>
 		<WarningLevel>4</WarningLevel>
 	</PropertyGroup>
 
diff --git a/src/GraphQL.Client/GraphQL.Client.csproj b/src/GraphQL.Client/GraphQL.Client.csproj
index 572356d7..8043bf35 100644
--- a/src/GraphQL.Client/GraphQL.Client.csproj
+++ b/src/GraphQL.Client/GraphQL.Client.csproj
@@ -5,7 +5,7 @@
 
 	<PropertyGroup>
 		<Description>A GraphQL Client</Description>
-		<TargetFramework>netstandard2.0</TargetFramework>
+		<TargetFrameworks>netstandard2.0</TargetFrameworks>
 	</PropertyGroup>
 
 	<ItemGroup>
@@ -14,6 +14,14 @@
 
 	<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" />
+	</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>
diff --git a/src/GraphQL.Client/Http/GraphQLHttpClient.cs b/src/GraphQL.Client/Http/GraphQLHttpClient.cs
index fe7bd7e7..f05cd2ef 100644
--- a/src/GraphQL.Client/Http/GraphQLHttpClient.cs
+++ b/src/GraphQL.Client/Http/GraphQLHttpClient.cs
@@ -1,6 +1,9 @@
 using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.Net.Http;
 using System.Net.Http.Headers;
+using System.Net.WebSockets;
 using System.Threading;
 using System.Threading.Tasks;
 using GraphQL.Client.Http.Internal;
@@ -37,9 +40,15 @@ public GraphQLHttpClientOptions Options {
 			set => this.graphQLHttpHandler.Options = value;
 		}
 
+		/// <inheritdoc />
+		[Obsolete("EXPERIMENTAL")]
+		public IObservable<Exception> WebSocketReceiveErrors => graphQlHttpWebSocket.ReceiveErrors;
+
 		#endregion
 
 		internal readonly GraphQLHttpHandler graphQLHttpHandler;
+		internal readonly GraphQLHttpWebSocket graphQlHttpWebSocket;
+		private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
 
 		/// <summary>
 		/// Initializes a new instance
@@ -66,14 +75,16 @@ public GraphQLHttpClient(string endPoint, GraphQLHttpClientOptions options) : th
 		/// <param name="endPoint">The EndPoint to be used</param>
 		/// <param name="options">The Options to be used</param>
 		public GraphQLHttpClient(Uri endPoint, GraphQLHttpClientOptions options) {
+
+			options.EndPoint = endPoint ?? throw new ArgumentNullException(nameof(endPoint));
+
 			if (options == null) { throw new ArgumentNullException(nameof(options)); }
-			if (options.EndPoint == null) { throw new ArgumentNullException(nameof(options.EndPoint)); }
 			if (options.JsonSerializerSettings == null) { throw new ArgumentNullException(nameof(options.JsonSerializerSettings)); }
 			if (options.HttpMessageHandler == null) { throw new ArgumentNullException(nameof(options.HttpMessageHandler)); }
 			if (options.MediaType == null) { throw new ArgumentNullException(nameof(options.MediaType)); }
 
-			options.EndPoint = endPoint ?? throw new ArgumentNullException(nameof(endPoint));
 			this.graphQLHttpHandler = new GraphQLHttpHandler(options);
+			this.graphQlHttpWebSocket = new GraphQLHttpWebSocket(_getWebSocketUri(), options);
 		}
 
 		/// <summary>
@@ -88,6 +99,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) {
@@ -98,42 +110,129 @@ 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) =>
 			this.SendQueryAsync(new GraphQLRequest(query), cancellationToken);
 
-		public Task<GraphQLResponse> SendQueryAsync(GraphQLRequest request, CancellationToken cancellationToken = default) =>
-			this.graphQLHttpHandler.PostAsync(request, cancellationToken);
+		public async Task<GraphQLResponse> SendQueryAsync(GraphQLRequest request, CancellationToken cancellationToken = default)
+		{
+			var preprocessed = await Options.PreprocessRequest(request, this).ConfigureAwait(false);
+
+			return Options.UseWebSocketForQueriesAndMutations
+				? await this.graphQlHttpWebSocket.Request(preprocessed, cancellationToken).ConfigureAwait(false)
+				: await this.graphQLHttpHandler.PostAsync(preprocessed, cancellationToken).ConfigureAwait(false);
+		}
 
 		public Task<GraphQLResponse> SendMutationAsync(string query, CancellationToken cancellationToken = default) =>
 			this.SendMutationAsync(new GraphQLRequest(query), cancellationToken);
 
 		public Task<GraphQLResponse> SendMutationAsync(GraphQLRequest request, CancellationToken cancellationToken = default) =>
-			this.graphQLHttpHandler.PostAsync(request, cancellationToken);
+			this.SendQueryAsync(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) {
+		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 preprocessed = Options.PreprocessRequest(request, this).GetAwaiter().GetResult();
 
-			var webSocketSchema = this.EndPoint.Scheme == "https" ? "wss" : "ws";
-			var webSocketUri = new Uri($"{webSocketSchema}://{this.EndPoint.Host}:{this.EndPoint.Port}{this.EndPoint.AbsolutePath}");
-			var graphQLSubscriptionResult = new GraphQLHttpSubscriptionResult(webSocketUri, request);
+			var graphQLSubscriptionResult = new GraphQLHttpSubscriptionResult(_getWebSocketUri(), preprocessed);
 			graphQLSubscriptionResult.StartAsync(cancellationToken);
-			return Task.FromResult<IGraphQLSubscriptionResult>(graphQLSubscriptionResult);
+			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];
+
+			var observable = graphQlHttpWebSocket.CreateSubscriptionStream(request, this, cancellationToken: _cancellationTokenSource.Token);
+
+			subscriptionStreams.TryAdd(request, observable);
+			return observable;
+		}
+
+		/// <inheritdoc />
+		[Obsolete("EXPERIMENTAL API")]
+		public IObservable<GraphQLResponse> CreateSubscriptionStream(GraphQLRequest request, Action<WebSocketException> webSocketExceptionHandler)
+		{
+			if (_disposed)
+				throw new ObjectDisposedException(nameof(GraphQLHttpClient));
+					   
+			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)
+		{
+			if (_disposed)
+				throw new ObjectDisposedException(nameof(GraphQLHttpClient));
+
+			if (subscriptionStreams.ContainsKey(request))
+				return subscriptionStreams[request];
+
+			var observable = graphQlHttpWebSocket.CreateSubscriptionStream(request, this, 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>
-		public void Dispose() =>
-			this.graphQLHttpHandler.Dispose();
+		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 73a00430..46d08abf 100644
--- a/src/GraphQL.Client/Http/GraphQLHttpClientOptions.cs
+++ b/src/GraphQL.Client/Http/GraphQLHttpClientOptions.cs
@@ -2,6 +2,9 @@
 using System.Collections.Generic;
 using System.Net.Http;
 using System.Net.Http.Headers;
+using System.Net.WebSockets;
+using System.Threading.Tasks;
+using GraphQL.Common.Request;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Converters;
 using Newtonsoft.Json.Serialization;
@@ -39,6 +42,25 @@ 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());
+		};
+
+		/// <summary>
+		/// If <see langword="true"/>, the websocket connection is also used for regular queries and mutations
+		/// </summary>
+		public bool UseWebSocketForQueriesAndMutations { get; set; } = false;
+
+		/// <summary>
+		/// Request preprocessing function. Can be used i.e. to inject authorization info into a GraphQL request payload.
+		/// </summary>
+		public Func<GraphQLRequest, GraphQLHttpClient, Task<GraphQLRequest>> PreprocessRequest { get; set; }= (request, client) => Task.FromResult(request);
 	}
 
 }
diff --git a/src/GraphQL.Client/Http/GraphQLHttpSubscriptionHelpers.cs b/src/GraphQL.Client/Http/GraphQLHttpSubscriptionHelpers.cs
new file mode 100644
index 00000000..24b3e32c
--- /dev/null
+++ b/src/GraphQL.Client/Http/GraphQLHttpSubscriptionHelpers.cs
@@ -0,0 +1,219 @@
+using System;
+using System.Diagnostics;
+using System.Net.WebSockets;
+using System.Reactive.Disposables;
+using System.Reactive.Linq;
+using System.Reactive.Threading.Tasks;
+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(
+			this GraphQLHttpWebSocket graphQlHttpWebSocket,
+			GraphQLRequest request,
+			GraphQLHttpClient client,
+			Action<Exception> exceptionHandler = null,
+			CancellationToken cancellationToken = default)
+		{
+			return Observable.Defer(() =>
+				Observable.Create<GraphQLResponse>(async observer =>
+				{
+					var preprocessedRequest = await client.Options.PreprocessRequest(request, client);
+					var startRequest = new GraphQLWebSocketRequest
+					{
+						Id = Guid.NewGuid().ToString("N"),
+						Type = GQLWebSocketMessageType.GQL_START,
+						Payload = preprocessedRequest
+					};
+					var closeRequest = new GraphQLWebSocketRequest
+					{
+						Id = startRequest.Id,
+						Type = GQLWebSocketMessageType.GQL_STOP
+					};
+
+					var observable = Observable.Create<GraphQLResponse>(o =>
+						graphQlHttpWebSocket.ResponseStream.Subscribe(response =>
+							{
+								// ignore null values and messages for other requests
+								if (response == null || response.Id != startRequest.Id) return;
+
+								// terminate the sequence when a 'complete' message is received
+								if (response.Type == GQLWebSocketMessageType.GQL_COMPLETE)
+								{
+									Debug.WriteLine($"received 'complete' message on subscription {startRequest.Id}");
+									o.OnCompleted();
+									return;
+								}
+
+								// post the GraphQLResponse to the stream (even if a GraphQL error occurred)
+								Debug.WriteLine($"received payload on subscription {startRequest.Id}");
+								o.OnNext(((JObject)response.Payload)?.ToObject<GraphQLResponse>());
+
+								// in case of a GraphQL error, terminate the sequence after the response has been posted
+								if (response.Type == GQLWebSocketMessageType.GQL_ERROR)
+								{
+									Debug.WriteLine($"terminating subscription {startRequest.Id} because of a GraphQL error");
+									o.OnCompleted();
+								}
+							},
+							o.OnError,
+							o.OnCompleted)
+					);
+
+					try
+					{
+						// initialize 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 (exceptionHandler == null) {
+							// if the external handler is not set, propagate all exceptions except WebSocketExceptions
+							// this will ensure that the client tries to re-establish subscriptions on connection loss
+							if (!(e is WebSocketException)) throw e;
+						}
+						else
+						{
+							// 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();
+		}
+
+		internal static Task<GraphQLResponse> Request(
+			this GraphQLHttpWebSocket graphQlHttpWebSocket,
+			GraphQLRequest request,
+			CancellationToken cancellationToken = default)
+		{
+			return Observable.Create<GraphQLResponse>(async observer =>
+			{
+				var websocketRequest = new GraphQLWebSocketRequest
+				{
+					Id = Guid.NewGuid().ToString("N"),
+					Type = GQLWebSocketMessageType.GQL_START,
+					Payload = request
+				};
+				var observable = graphQlHttpWebSocket.ResponseStream
+					.Where(response => response != null && response.Id == websocketRequest.Id)
+					.TakeUntil(response => response.Type == GQLWebSocketMessageType.GQL_COMPLETE)
+					.Select(response =>
+					{
+						Debug.WriteLine($"received response for request {websocketRequest.Id}");
+						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)
+				);
+
+				Debug.WriteLine($"submitting request {websocketRequest.Id}");
+				// send request
+				try
+				{
+					await graphQlHttpWebSocket.SendWebSocketRequest(websocketRequest).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>())
+			.FirstOrDefaultAsync()
+			.ToTask(cancellationToken);
+		}
+	}
+}
diff --git a/src/GraphQL.Client/Http/GraphQLHttpSubscriptionResult.cs b/src/GraphQL.Client/Http/GraphQLHttpSubscriptionResult.cs
index c916f82e..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,10 +39,12 @@ 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);
-					if (webSocketResponse != null) {
-						this.LastResponse = webSocketResponse.Payload;
-						this.OnReceive?.Invoke(webSocketResponse.Payload);
+					var webSocketResponse = JsonConvert.DeserializeObject<GraphQLWebSocketResponse>(stringResult);
+					if (webSocketResponse != null)
+					{
+						var response = (GraphQLResponse) webSocketResponse.Payload;
+						this.LastResponse = response;
+						this.OnReceive?.Invoke(response);
 					}
 				}
 			}
@@ -60,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
@@ -69,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
@@ -77,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..eb2881ba
--- /dev/null
+++ b/src/GraphQL.Client/Http/GraphQLHttpWebSocket.cs
@@ -0,0 +1,309 @@
+using System;
+using System.Diagnostics;
+using System.IO;
+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 ArraySegment<byte> buffer;
+		private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
+
+		private Subject<GraphQLWebSocketResponse> _responseSubject;
+		private Subject<GraphQLWebSocketRequest> _requestSubject = new Subject<GraphQLWebSocketRequest>();
+		private Subject<Exception> _exceptionSubject = new Subject<Exception>();
+		private IDisposable _requestSubscription;
+
+		public WebSocketState WebSocketState => clientWebSocket?.State ?? WebSocketState.None;
+
+		private WebSocket clientWebSocket = null;
+		private int _connectionAttempt = 0;
+
+		public GraphQLHttpWebSocket(Uri webSocketUri, GraphQLHttpClientOptions options)
+		{
+			this.webSocketUri = webSocketUri;
+			_options = options;
+			buffer = new ArraySegment<byte>(new byte[8192]);
+			_responseStream = _createResponseStream();
+
+			_requestSubscription = _requestSubject.Select(request => Observable.FromAsync(() => _sendWebSocketRequest(request))).Concat().Subscribe();
+		}
+
+		public IObservable<Exception> ReceiveErrors => _exceptionSubject.AsObservable();
+
+		public IObservable<GraphQLWebSocketResponse> ResponseStream => _responseStream;
+		public IObservable<GraphQLWebSocketResponse> _responseStream;
+		//private IDisposable _responseStreamConnection;
+
+		public Task SendWebSocketRequest(GraphQLWebSocketRequest request)
+		{
+			_requestSubject.OnNext(request);
+			return request.SendTask();
+		}
+
+		private async Task _sendWebSocketRequest(GraphQLWebSocketRequest request)
+		{
+			try
+			{
+				if (_cancellationTokenSource.Token.IsCancellationRequested)
+				{
+					request.SendCanceled();
+					return;
+				}
+
+				await InitializeWebSocket().ConfigureAwait(false);
+				var webSocketRequestString = JsonConvert.SerializeObject(request, _options.JsonSerializerSettings);
+				await this.clientWebSocket.SendAsync(
+					new ArraySegment<byte>(Encoding.UTF8.GetBytes(webSocketRequestString)),
+					WebSocketMessageType.Text,
+					true,
+					_cancellationTokenSource.Token).ConfigureAwait(false);
+				request.SendCompleted();
+			}
+			catch (Exception e)
+			{
+				request.SendFailed(e);
+			}
+		}
+
+		public Task InitializeWebSocketTask { get; private set; } = Task.CompletedTask;
+
+		private object _initializeLock = new object();
+
+		#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();
+
+			lock (_initializeLock)
+			{
+				// if an initialization task is already running, return that
+				if(InitializeWebSocketTask != null &&
+				   !InitializeWebSocketTask.IsFaulted &&
+				   !InitializeWebSocketTask.IsCompleted)
+					return InitializeWebSocketTask;
+
+				// if the websocket is open, return a completed task
+				if (clientWebSocket != null && clientWebSocket.State == WebSocketState.Open)
+					return Task.CompletedTask;
+
+				// else (re-)create websocket and connect
+				//_responseStreamConnection?.Dispose();
+				clientWebSocket?.Dispose();
+
+				// fix websocket not supported on win 7 using
+				// https://github.com/PingmanTools/System.Net.WebSockets.Client.Managed
+				clientWebSocket = SystemClientWebSocket.CreateClientWebSocket();
+				switch (clientWebSocket)
+				{
+					case ClientWebSocket nativeWebSocket:
+						nativeWebSocket.Options.AddSubProtocol("graphql-ws");
+						break;
+					case System.Net.WebSockets.Managed.ClientWebSocket managedWebSocket:
+						managedWebSocket.Options.AddSubProtocol("graphql-ws");
+						break;
+					default:
+						throw new NotSupportedException($"unknown websocket type {clientWebSocket.GetType().Name}");
+				}
+
+				return InitializeWebSocketTask = _connectAsync(_cancellationTokenSource.Token);
+			}
+		}
+		
+		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)
+		{
+			if (_responseSubject == null || _responseSubject.IsDisposed)
+			{
+				_responseSubject = new Subject<GraphQLWebSocketResponse>();
+				var observable = await _getReceiveResultStream().ConfigureAwait(false);
+				observable.Subscribe(_responseSubject);
+
+				_responseSubject.Subscribe(_ => { }, ex =>
+				{
+					_exceptionSubject.OnNext(ex);
+					_responseSubject?.Dispose();
+					_responseSubject = null;
+				},
+				() => {
+					_responseSubject?.Dispose();
+					_responseSubject = null;
+				});
+			}
+					   
+			return new CompositeDisposable
+			(
+				_responseSubject.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)
+		{
+			try
+			{
+				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()}");
+				//_responseStreamConnection = _responseStream.Connect();
+				_connectionAttempt = 1;
+			}
+			catch (Exception e)
+			{
+				_exceptionSubject.OnNext(e);
+				throw;
+			}
+		}
+
+
+		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 data on websocket {clientWebSocket.GetHashCode()} ...");
+				WebSocketReceiveResult webSocketReceiveResult = null;
+
+				using (var ms = new MemoryStream())
+				{
+					do
+					{
+						_cancellationTokenSource.Token.ThrowIfCancellationRequested();
+						webSocketReceiveResult = await clientWebSocket.ReceiveAsync(buffer, CancellationToken.None);
+						ms.Write(buffer.Array, buffer.Offset, webSocketReceiveResult.Count);
+					}
+					while (!webSocketReceiveResult.EndOfMessage);
+
+					_cancellationTokenSource.Token.ThrowIfCancellationRequested();
+					ms.Seek(0, SeekOrigin.Begin);
+
+					if (webSocketReceiveResult.MessageType == WebSocketMessageType.Text)
+					{
+						using (var reader = new StreamReader(ms, Encoding.UTF8))
+						{
+							var stringResult = await reader.ReadToEndAsync();
+							Debug.WriteLine($"data received on websocket {clientWebSocket.GetHashCode()}: {stringResult}");
+							return JsonConvert.DeserializeObject<GraphQLWebSocketResponse>(stringResult, _options.JsonSerializerSettings);
+						}
+					}
+					else
+					{
+						throw new NotSupportedException("binary websocket messages are not supported");
+					}
+				}				
+			}
+			catch (Exception e)
+			{
+				Debug.WriteLine($"exception thrown while receiving websocket data: {e}");
+				throw;
+			}
+		}
+
+		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/Http/HttpClientExtensions.cs b/src/GraphQL.Client/Http/HttpClientExtensions.cs
index c0fb4b49..1d7ca0e4 100644
--- a/src/GraphQL.Client/Http/HttpClientExtensions.cs
+++ b/src/GraphQL.Client/Http/HttpClientExtensions.cs
@@ -1,3 +1,4 @@
+using System;
 using System.Net.Http;
 
 namespace GraphQL.Client.Http {
@@ -16,6 +17,15 @@ public static class HttpClientExtensions {
 		public static GraphQLHttpClient AsGraphQLClient(this HttpClient httpClient, GraphQLHttpClientOptions graphQLHttpClientOptions) =>
 			new GraphQLHttpClient(graphQLHttpClientOptions, httpClient);
 
+		/// <summary>
+		/// Creates a <see cref="GraphQLHttpClient"/> from a <see cref="HttpClient"/>
+		/// </summary>
+		/// <param name="httpClient"></param>
+		/// <param name="endPoint"></param>
+		/// <returns></returns>
+		public static GraphQLHttpClient AsGraphQLClient(this HttpClient httpClient, Uri endPoint) =>
+			new GraphQLHttpClient(new GraphQLHttpClientOptions { EndPoint = endPoint }, httpClient);
+
 	}
 
 }
diff --git a/src/GraphQL.Client/IGraphQLClient.cs b/src/GraphQL.Client/IGraphQLClient.cs
index 108762f7..098d0524 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,6 +37,47 @@ 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);
+
+		/// <summary>
+		/// Publishes all exceptions which occur inside the websocket receive stream (i.e. for logging purposes)
+		/// </summary>
+		[Obsolete("EXPERIMENTAL")]
+		IObservable<Exception> WebSocketReceiveErrors { get; }
 	}
 
 }
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/GraphQLRequest.cs b/src/GraphQL.Common/Request/GraphQLRequest.cs
index 229722dd..d04f1992 100644
--- a/src/GraphQL.Common/Request/GraphQLRequest.cs
+++ b/src/GraphQL.Common/Request/GraphQLRequest.cs
@@ -33,6 +33,15 @@ public GraphQLRequest(string query){
 			this.Query = query;
 		}
 
+		/// <summary>
+		/// Initialize a new GraphQLRequest
+		/// </summary>
+		[Obsolete("Pass query as parameter! This constructor will be removed in a future version!")]
+		public GraphQLRequest()
+		{
+			this.Query = string.Empty;
+		}
+
 		/// <inheritdoc />
 		public override bool Equals(object? obj) => this.Equals(obj as GraphQLRequest);
 
@@ -47,11 +56,13 @@ public bool Equals(GraphQLRequest? other) {
 		}
 
 		/// <inheritdoc />
-		public override int GetHashCode(){
-			unchecked{
-				var hashCode = EqualityComparer<string>.Default.GetHashCode(this.Query);
-				hashCode = (hashCode * 397) ^ EqualityComparer<string?>.Default.GetHashCode(this.OperationName);
-				hashCode = (hashCode * 397) ^ EqualityComparer<dynamic?>.Default.GetHashCode(this.Variables);
+		public override int GetHashCode()
+		{
+			unchecked
+			{
+				var hashCode = EqualityComparer<string?>.Default.GetHashCode(Query);
+				hashCode = (hashCode * 397) ^ EqualityComparer<string?>.Default.GetHashCode(OperationName);
+				hashCode = (hashCode * 397) ^ EqualityComparer<dynamic?>.Default.GetHashCode(Variables);
 				return hashCode;
 			}
 		}
diff --git a/src/GraphQL.Common/Request/GraphQLSubscriptionRequest.cs b/src/GraphQL.Common/Request/GraphQLWebSocketRequest.cs
similarity index 50%
rename from src/GraphQL.Common/Request/GraphQLSubscriptionRequest.cs
rename to src/GraphQL.Common/Request/GraphQLWebSocketRequest.cs
index a468a0b3..d805efc4 100644
--- a/src/GraphQL.Common/Request/GraphQLSubscriptionRequest.cs
+++ b/src/GraphQL.Common/Request/GraphQLWebSocketRequest.cs
@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading.Tasks;
 
 namespace GraphQL.Common.Request {
 
@@ -7,7 +8,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
@@ -24,11 +25,35 @@ public class GraphQLSubscriptionRequest : IEquatable<GraphQLSubscriptionRequest>
 		/// </summary>
 		public GraphQLRequest Payload { get; set; }
 
+		private TaskCompletionSource<bool> _tcs = new TaskCompletionSource<bool>();
+
+		/// <summary>
+		/// Task used to await the actual send operation and to convey potential exceptions
+		/// </summary>
+		/// <returns></returns>
+		public Task SendTask() => _tcs.Task;
+
+		/// <summary>
+		/// gets called when the send operation for this request has completed sucessfully
+		/// </summary>
+		public void SendCompleted() => _tcs.SetResult(true);
+
+		/// <summary>
+		/// gets called when an exception occurs during the send operation
+		/// </summary>
+		/// <param name="e"></param>
+		public void SendFailed(Exception e) => _tcs.SetException(e);
+
+		/// <summary>
+		/// gets called when the GraphQLHttpWebSocket has been disposed before the send operation for this request has started
+		/// </summary>
+		public void SendCanceled() => _tcs.SetCanceled();
+
 		/// <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 +82,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 70%
rename from src/GraphQL.Common/Response/GraphQLSubscriptionResponse.cs
rename to src/GraphQL.Common/Response/GraphQLWebSocketResponse.cs
index 3ba69331..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
@@ -22,13 +22,13 @@ public class GraphQLSubscriptionResponse : IEquatable<GraphQLSubscriptionRespons
 		/// <summary>
 		/// The Payload of the Response
 		/// </summary>
-		public GraphQLResponse Payload { get; set; }
+		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.Client.Tests/GraphQL.Client.Tests.csproj b/tests/GraphQL.Client.Tests/GraphQL.Client.Tests.csproj
index 69a8d9da..1c9a24d9 100644
--- a/tests/GraphQL.Client.Tests/GraphQL.Client.Tests.csproj
+++ b/tests/GraphQL.Client.Tests/GraphQL.Client.Tests.csproj
@@ -1,4 +1,4 @@
-<?xml version="1.0" encoding="utf-8"?>
+<?xml version="1.0" encoding="utf-8"?>
 <Project Sdk="Microsoft.NET.Sdk">
 
 	<Import Project="../tests.props" />
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/GraphQL.Integration.Tests.csproj b/tests/GraphQL.Integration.Tests/GraphQL.Integration.Tests.csproj
new file mode 100644
index 00000000..5fc195bc
--- /dev/null
+++ b/tests/GraphQL.Integration.Tests/GraphQL.Integration.Tests.csproj
@@ -0,0 +1,28 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp2.1</TargetFramework>
+
+    <IsPackable>false</IsPackable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.AspNetCore.App" Version="2.1.5" />
+    <PackageReference Include="Microsoft.AspNetCore.Mvc.Testing" Version="2.1.3" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.9.0" />
+    <PackageReference Include="System.Reactive" Version="4.1.2" />
+    <PackageReference Include="System.Reactive.Compatibility" Version="4.1.2" />
+    <PackageReference Include="xunit" Version="2.4.1" />
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.1">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\GraphQL.Client\GraphQL.Client.csproj" />
+    <ProjectReference Include="..\..\src\GraphQL.Common\GraphQL.Common.csproj" />
+    <ProjectReference Include="..\IntegrationTestServer\IntegrationTestServer.csproj" />
+  </ItemGroup>
+
+</Project>
diff --git a/tests/GraphQL.Integration.Tests/GraphQLClientExtensions.cs b/tests/GraphQL.Integration.Tests/GraphQLClientExtensions.cs
new file mode 100644
index 00000000..ff71ee6d
--- /dev/null
+++ b/tests/GraphQL.Integration.Tests/GraphQLClientExtensions.cs
@@ -0,0 +1,51 @@
+using System;
+using System.Threading.Tasks;
+using GraphQL.Client.Http;
+using GraphQL.Common.Request;
+using GraphQL.Common.Response;
+
+namespace GraphQL.Integration.Tests
+{
+	public static class GraphQLClientExtensions
+	{
+		public static async Task<GraphQLResponse> AddMessageAsync(this GraphQLHttpClient client, string message)
+		{
+			var graphQLRequest = new GraphQLRequest(@"
+				mutation($input: MessageInputType){
+				  addMessage(message: $input){
+				    content
+				  }
+				}")
+			{
+				Variables = new
+				{
+					input = new
+					{
+						fromId = "2",
+						content = message,
+						sentAt = DateTime.Now
+					}
+				}
+			};
+			return await client.SendMutationAsync(graphQLRequest).ConfigureAwait(false);
+		}
+
+		public static async Task<GraphQLResponse> JoinDeveloperUser(this GraphQLHttpClient client)
+		{
+			var graphQLRequest = new GraphQLRequest(@"
+				mutation($userId: String){
+				  join(userId: $userId){
+				    displayName
+				    id
+				  }
+				}")
+			{
+				Variables = new
+				{
+					userId = "1"
+				}
+			};
+			return await client.SendMutationAsync(graphQLRequest).ConfigureAwait(false);
+		}
+	}
+}
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
new file mode 100644
index 00000000..1569176d
--- /dev/null
+++ b/tests/GraphQL.Integration.Tests/ObservableTester.cs
@@ -0,0 +1,152 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace GraphQL.Integration.Tests
+{
+	public class ObservableTester<T> : IDisposable
+	{
+		private readonly IDisposable _subscription;
+		private ManualResetEventSlim _updateReceived { get; } = new ManualResetEventSlim();
+		private ManualResetEventSlim _completed { get; } = new ManualResetEventSlim();
+		private ManualResetEventSlim _error { 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 UpdateReceived => _updateReceived.IsSet;
+		/// <summary>
+		/// The last payload which was received.
+		/// </summary>
+		public T LastPayload { get; private set; }
+
+		public Exception Error { get; private set; }
+
+		/// <summary>
+		/// Creates a new <see cref="ObservableTester{T}"/> which subscribes to the supplied <see cref="IObservable{T}"/>
+		/// </summary>
+		/// <param name="observable">the <see cref="IObservable{T}"/> under test</param>
+		public ObservableTester(IObservable<T> observable)
+		{
+			_subscription = observable.Subscribe(
+				obj => {
+					LastPayload = obj;
+					_updateReceived.Set();
+				},
+				ex =>
+				{
+					Error = ex;
+					_error.Set();
+				},
+				() => _completed.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 ShouldHaveReceivedUpdate(Action<T> assertPayload = null, TimeSpan? timeout = null)
+		{
+			try
+			{
+				if (!_updateReceived.Wait(timeout ?? Timeout))
+					Assert.True(false, $"no update received 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 ShouldNotHaveReceivedUpdate(TimeSpan? timeout = null)
+		{
+			if(!timeout.HasValue) timeout = TimeSpan.FromMilliseconds(100);
+			try
+			{
+				if (_updateReceived.Wait(timeout.Value))
+					Assert.True(false, "update was inadvertently pushed!");
+			}
+			finally
+			{
+				_reset();
+			}
+		}
+
+		/// <summary>
+		/// Asserts that the subscription has completed within the configured <see cref="Timeout"/> since the last <see cref="_reset"/>
+		/// </summary>
+		public void ShouldHaveCompleted(TimeSpan? timeout = null)
+		{
+			try
+			{
+				if (!_completed.Wait(timeout ?? Timeout))
+					Assert.True(false, $"subscription did not complete within {(timeout ?? Timeout).TotalSeconds} s!");
+			}
+			finally
+			{
+				_reset();
+			}
+		}
+
+		/// <summary>
+		/// 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, TimeSpan? timeout = null)
+		{
+			try
+			{
+				if (!_error.Wait(timeout ?? Timeout))
+					Assert.True(false, $"subscription did not throw an error within {(timeout ?? Timeout).TotalSeconds} s!");
+
+				assertError?.Invoke(Error);
+			}
+			finally
+			{
+				_reset();
+			}
+		}
+
+		/// <summary>
+		/// Resets the tester class. Should be called before triggering the potential update
+		/// </summary>
+		private void _reset()
+		{
+			//if (_completed.IsSet)
+			//	throw new InvalidOperationException(
+			//		"the subscription sequence has completed. this tester instance cannot be reused");
+
+			LastPayload = default(T);
+			_updateReceived.Reset();
+		}
+
+		/// <inheritdoc />
+		public void Dispose()
+		{
+			_subscription?.Dispose();
+		}
+	}
+
+	public static class ObservableExtensions
+	{
+		public static ObservableTester<T> SubscribeTester<T>(this IObservable<T> observable)
+		{
+			return new ObservableTester<T>(observable);
+		}
+	}
+}
diff --git a/tests/GraphQL.Integration.Tests/Properties/launchSettings.json b/tests/GraphQL.Integration.Tests/Properties/launchSettings.json
new file mode 100644
index 00000000..f8fea71b
--- /dev/null
+++ b/tests/GraphQL.Integration.Tests/Properties/launchSettings.json
@@ -0,0 +1,27 @@
+{
+  "iisSettings": {
+    "windowsAuthentication": false,
+    "anonymousAuthentication": true,
+    "iisExpress": {
+      "applicationUrl": "http://localhost:51432/",
+      "sslPort": 0
+    }
+  },
+  "profiles": {
+    "IIS Express": {
+      "commandName": "IISExpress",
+      "launchBrowser": true,
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    },
+    "GraphQL.Integration.Tests": {
+      "commandName": "Project",
+      "launchBrowser": true,
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      },
+      "applicationUrl": "http://localhost:51433/"
+    }
+  }
+}
\ No newline at end of file
diff --git a/tests/GraphQL.Integration.Tests/SubscriptionsTest.cs b/tests/GraphQL.Integration.Tests/SubscriptionsTest.cs
new file mode 100644
index 00000000..fb704ad4
--- /dev/null
+++ b/tests/GraphQL.Integration.Tests/SubscriptionsTest.cs
@@ -0,0 +1,363 @@
+using System;
+using System.Diagnostics;
+using System.Net.WebSockets;
+using System.Threading;
+using System.Threading.Tasks;
+using GraphQL.Client.Http;
+using GraphQL.Common.Request;
+using GraphQL.Common.Response;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Logging;
+using Xunit;
+
+namespace GraphQL.Integration.Tests
+{
+	public class SubscriptionsTest
+	{
+		public static IWebHost CreateServer(int port)
+		{
+			var configBuilder = new ConfigurationBuilder();
+			configBuilder.AddInMemoryCollection();
+			var config = configBuilder.Build();
+			config["server.urls"] = $"http://localhost:{port}";
+
+			var host = new WebHostBuilder()
+				.ConfigureLogging((ctx, logging) =>
+				{
+					logging.AddDebug();
+				})
+				.UseConfiguration(config)
+				.UseKestrel()
+				.UseStartup<IntegrationTestServer.Startup>()
+				.Build();
+
+			host.Start();
+
+			return host;
+		}
+
+		private readonly IWebHost _server;
+
+		public SubscriptionsTest()
+		{
+		}
+
+		private GraphQLHttpClient GetGraphQLClient(int port, bool requestsViaWebsocket = false)
+			=> new GraphQLHttpClient(new GraphQLHttpClientOptions
+			{
+				EndPoint = new Uri($"http://localhost:{port}/graphql"),
+				UseWebSocketForQueriesAndMutations = requestsViaWebsocket
+			});
+
+
+		[Fact]
+		public async void AssertTestingHarness()
+		{
+			var port = NetworkHelpers.GetFreeTcpPortNumber();
+			using (CreateServer(port))
+			{
+				var client = GetGraphQLClient(port);
+
+				const string message = "some random testing message";
+				var response = await client.AddMessageAsync(message).ConfigureAwait(false);
+
+				Assert.Equal(message, (string) response.Data.addMessage.content);
+			}
+		}
+
+		[Fact]
+		public async void CanSendRequestViaWebsocket()
+		{
+			var port = NetworkHelpers.GetFreeTcpPortNumber();
+			using (CreateServer(port))
+			{
+				var client = GetGraphQLClient(port, true);
+				const string message = "some random testing message";
+				var response = await client.AddMessageAsync(message).ConfigureAwait(false);
+
+				Assert.Equal(message, (string)response.Data.addMessage.content);
+			}
+		}
+
+		[Fact]
+		public async void CanHandleRequestErrorViaWebsocket()
+		{
+			var port = NetworkHelpers.GetFreeTcpPortNumber();
+			using (CreateServer(port))
+			{
+				var client = GetGraphQLClient(port, true);
+				const string message = "some random testing message";
+				var response = await client.SendQueryAsync(new GraphQLRequest("this query is formatted quite badly")).ConfigureAwait(false);
+
+				Assert.Single(response.Errors);
+			}
+		}
+
+		private const string SubscriptionQuery = @"
+			subscription {
+			  messageAdded{
+			    content
+			  }
+			}";
+
+		private readonly GraphQLRequest SubscriptionRequest = new GraphQLRequest(SubscriptionQuery);
+
+		[Fact]
+		public async void CanCreateObservableSubscription()
+		{
+			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 = "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 {
+			  userJoined{
+				displayName
+				id
+			  }
+			}";
+
+		private readonly GraphQLRequest SubscriptionRequest2 = new GraphQLRequest(SubscriptionQuery2);
+
+		[Fact]
+		public async void CanConnectTwoSubscriptionsSimultaneously()
+		{
+			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> observable1 = client.CreateSubscriptionStream(SubscriptionRequest, callbackTester.Callback);
+				IObservable<GraphQLResponse> observable2 = client.CreateSubscriptionStream(SubscriptionRequest2, callbackTester2.Callback);
+
+				Debug.WriteLine("subscribing...");
+				var tester = observable1.SubscribeTester();
+				var tester2 = observable2.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);
+				});
+
+				await Task.Delay(500); // ToDo: can be removed after https://github.com/graphql-dotnet/server/pull/199 was merged and released
+
+				response = await client.JoinDeveloperUser().ConfigureAwait(false);
+				Assert.Equal("developer", (string)response.Data.join.displayName.Value);
+
+				tester2.ShouldHaveReceivedUpdate(gqlResponse =>
+				{
+					Assert.Equal("1", (string)gqlResponse.Data.userJoined.id.Value);
+					Assert.Equal("developer", (string)gqlResponse.Data.userJoined.displayName.Value);
+				});
+				
+				Debug.WriteLine("disposing subscription...");
+				tester2.Dispose();
+
+				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();
+			}
+		}
+
+		[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();
+		}
+		
+		[Fact]
+		public async void CanHandleSubscriptionError()
+		{
+			var port = NetworkHelpers.GetFreeTcpPortNumber();
+			using (CreateServer(port))
+			{
+				var client = GetGraphQLClient(port);
+				Debug.WriteLine("creating subscription stream");
+				IObservable<GraphQLResponse> observable = client.CreateSubscriptionStream(
+					new GraphQLRequest(@"
+						subscription {
+						  failImmediately {
+						    content
+						  }
+						}")
+					);
+
+				Debug.WriteLine("subscribing...");
+				var tester = observable.SubscribeTester();
+				tester.ShouldHaveReceivedUpdate(gqlResponse =>
+				{
+					Assert.Single(gqlResponse.Errors);
+				});
+				tester.ShouldHaveCompleted();
+
+				client.Dispose();
+			}
+		}
+
+		[Fact]
+		public async void CanHandleQueryErrorInSubscription()
+		{
+			var port = NetworkHelpers.GetFreeTcpPortNumber();
+			using (CreateServer(port))
+			{
+				var client = GetGraphQLClient(port);
+				Debug.WriteLine("creating subscription stream");
+				IObservable<GraphQLResponse> observable = client.CreateSubscriptionStream(
+					new GraphQLRequest(@"
+						subscription {
+						  fieldDoesNotExist {
+						    content
+						  }
+						}")
+				);
+
+				Debug.WriteLine("subscribing...");
+				var tester = observable.SubscribeTester();
+				tester.ShouldHaveReceivedUpdate(gqlResponse =>
+				{
+					Assert.Single(gqlResponse.Errors);
+				});
+				tester.ShouldHaveCompleted();
+
+				client.Dispose();
+			}
+		}
+	}
+}
diff --git a/tests/GraphQL.Integration.Tests/WebApplicationFactoryExtensions.cs b/tests/GraphQL.Integration.Tests/WebApplicationFactoryExtensions.cs
new file mode 100644
index 00000000..32b32e61
--- /dev/null
+++ b/tests/GraphQL.Integration.Tests/WebApplicationFactoryExtensions.cs
@@ -0,0 +1,19 @@
+using System;
+using GraphQL.Client.Http;
+using Microsoft.AspNetCore.Mvc.Testing;
+
+namespace GraphQL.Integration.Tests
+{
+	public static class WebApplicationFactoryExtensions
+	{
+		public static GraphQLHttpClient CreateGraphQlHttpClient<TEntryPoint>(
+			this WebApplicationFactory<TEntryPoint> factory, string graphQlSchemaUrl, string urlScheme = "http://") where TEntryPoint: class
+		{
+			var httpClient = factory.CreateClient();
+			var uriBuilder = new UriBuilder(httpClient.BaseAddress);
+			uriBuilder.Path = graphQlSchemaUrl;
+			uriBuilder.Scheme = urlScheme;
+			return httpClient.AsGraphQLClient(uriBuilder.Uri);
+		}
+	}
+}
diff --git a/tests/IntegrationTestServer/ChatSchema/ChatMutation.cs b/tests/IntegrationTestServer/ChatSchema/ChatMutation.cs
new file mode 100644
index 00000000..0ca1dfbd
--- /dev/null
+++ b/tests/IntegrationTestServer/ChatSchema/ChatMutation.cs
@@ -0,0 +1,42 @@
+using GraphQL.Types;
+
+namespace IntegrationTestServer.ChatSchema
+{
+    public class ChatMutation : ObjectGraphType<object>
+    {
+        public ChatMutation(IChat chat)
+        {
+            Field<MessageType>("addMessage",
+                arguments: new QueryArguments(
+                    new QueryArgument<MessageInputType> {Name = "message"}
+                ),
+                resolve: context =>
+                {
+                    var receivedMessage = context.GetArgument<ReceivedMessage>("message");
+                    var message = chat.AddMessage(receivedMessage);
+                    return message;
+                });
+
+			Field<MessageFromType>("join",
+				arguments: new QueryArguments(
+					new QueryArgument<StringGraphType> { Name = "userId" }
+				),
+				resolve: context =>
+				{
+					var userId = context.GetArgument<string>("userId");
+					var userJoined = chat.Join(userId);
+					return userJoined;
+				});
+		}
+    }
+
+    public class MessageInputType : InputObjectGraphType
+    {
+        public MessageInputType()
+        {
+            Field<StringGraphType>("fromId");
+            Field<StringGraphType>("content");
+            Field<DateGraphType>("sentAt");
+        }
+    }
+}
diff --git a/tests/IntegrationTestServer/ChatSchema/ChatQuery.cs b/tests/IntegrationTestServer/ChatSchema/ChatQuery.cs
new file mode 100644
index 00000000..579c364f
--- /dev/null
+++ b/tests/IntegrationTestServer/ChatSchema/ChatQuery.cs
@@ -0,0 +1,13 @@
+using System.Linq;
+using GraphQL.Types;
+
+namespace IntegrationTestServer.ChatSchema
+{
+    public class ChatQuery : ObjectGraphType
+    {
+        public ChatQuery(IChat chat)
+        {
+            Field<ListGraphType<MessageType>>("messages", resolve: context => chat.AllMessages.Take(100));
+        }
+    }
+}
diff --git a/tests/IntegrationTestServer/ChatSchema/ChatSchema.cs b/tests/IntegrationTestServer/ChatSchema/ChatSchema.cs
new file mode 100644
index 00000000..904623a2
--- /dev/null
+++ b/tests/IntegrationTestServer/ChatSchema/ChatSchema.cs
@@ -0,0 +1,14 @@
+using GraphQL.Types;
+
+namespace IntegrationTestServer.ChatSchema
+{
+    public class ChatSchema : Schema
+    {
+        public ChatSchema(IChat chat) 
+        {
+            Query = new ChatQuery(chat);
+            Mutation = new ChatMutation(chat);
+            Subscription = new ChatSubscriptions(chat);
+        }
+    }
+}
diff --git a/tests/IntegrationTestServer/ChatSchema/ChatSubscriptions.cs b/tests/IntegrationTestServer/ChatSchema/ChatSubscriptions.cs
new file mode 100644
index 00000000..178d2b78
--- /dev/null
+++ b/tests/IntegrationTestServer/ChatSchema/ChatSubscriptions.cs
@@ -0,0 +1,99 @@
+using System;
+using System.Linq;
+using System.Reactive.Linq;
+using System.Security.Claims;
+using GraphQL;
+using GraphQL.Resolvers;
+using GraphQL.Server.Transports.Subscriptions.Abstractions;
+using GraphQL.Subscription;
+using GraphQL.Types;
+
+namespace IntegrationTestServer.ChatSchema
+{
+    public class ChatSubscriptions : ObjectGraphType<object>
+    {
+        private readonly IChat _chat;
+
+        public ChatSubscriptions(IChat chat)
+        {
+            _chat = chat;
+            AddField(new EventStreamFieldType
+            {
+                Name = "messageAdded",
+                Type = typeof(MessageType),
+                Resolver = new FuncFieldResolver<Message>(ResolveMessage),
+                Subscriber = new EventStreamResolver<Message>(Subscribe)
+            });
+
+			AddField(new EventStreamFieldType
+			{
+				Name = "contentAdded",
+				Type = typeof(MessageType),
+				Resolver = new FuncFieldResolver<Message>(ResolveMessage),
+				Subscriber = new EventStreamResolver<Message>(Subscribe)
+			});
+
+			AddField(new EventStreamFieldType
+            {
+                Name = "messageAddedByUser",
+                Arguments = new QueryArguments(
+                    new QueryArgument<NonNullGraphType<StringGraphType>> {Name = "id"}
+                ),
+                Type = typeof(MessageType),
+                Resolver = new FuncFieldResolver<Message>(ResolveMessage),
+                Subscriber = new EventStreamResolver<Message>(SubscribeById)
+            });
+
+			AddField(new EventStreamFieldType
+			{
+				Name = "userJoined",
+				Type = typeof(MessageFromType),
+				Resolver = new FuncFieldResolver<MessageFrom>(context => context.Source as MessageFrom),
+				Subscriber = new EventStreamResolver<MessageFrom>(context => _chat.UserJoined())
+			});
+
+
+			AddField(new EventStreamFieldType
+			{
+				Name = "failImmediately",
+				Type = typeof(MessageType),
+				Resolver = new FuncFieldResolver<Message>(ResolveMessage),
+				Subscriber = new EventStreamResolver<Message>(context => throw new NotSupportedException("this is supposed to fail"))
+			});
+		}
+
+        private IObservable<Message> SubscribeById(ResolveEventStreamContext context)
+        {
+            var messageContext = context.UserContext.As<MessageHandlingContext>();
+            var user = messageContext.Get<ClaimsPrincipal>("user");
+
+            var sub = "Anonymous";
+            if (user != null)
+                sub = user.Claims.FirstOrDefault(c => c.Type == "sub")?.Value;
+
+            var messages = _chat.Messages(sub);
+
+            var id = context.GetArgument<string>("id");
+            return messages.Where(message => message.From.Id == id);
+        }
+
+        private Message ResolveMessage(ResolveFieldContext context)
+        {
+            var message = context.Source as Message;
+
+            return message;
+        }
+
+        private IObservable<Message> Subscribe(ResolveEventStreamContext context)
+        {
+            var messageContext = context.UserContext.As<MessageHandlingContext>();
+            var user = messageContext.Get<ClaimsPrincipal>("user");
+
+            var sub = "Anonymous";
+            if (user != null)
+                sub = user.Claims.FirstOrDefault(c => c.Type == "sub")?.Value;
+
+            return _chat.Messages(sub);
+        }
+    }
+}
diff --git a/tests/IntegrationTestServer/ChatSchema/IChat.cs b/tests/IntegrationTestServer/ChatSchema/IChat.cs
new file mode 100644
index 00000000..e52c0432
--- /dev/null
+++ b/tests/IntegrationTestServer/ChatSchema/IChat.cs
@@ -0,0 +1,110 @@
+using System;
+using System.Collections.Concurrent;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+
+namespace IntegrationTestServer.ChatSchema
+{
+    public interface IChat
+    {
+        ConcurrentStack<Message> AllMessages { get; }
+
+        Message AddMessage(Message message);
+
+		MessageFrom Join(string userId);
+
+		IObservable<Message> Messages(string user);
+		IObservable<MessageFrom> UserJoined();
+
+		Message AddMessage(ReceivedMessage message);
+    }
+
+    public class Chat : IChat
+    {
+        private readonly ISubject<Message> _messageStream = new ReplaySubject<Message>(1);
+		private readonly ISubject<MessageFrom> _userJoined = new Subject<MessageFrom>();
+
+		public Chat()
+        {
+            AllMessages = new ConcurrentStack<Message>();
+            Users = new ConcurrentDictionary<string, string>
+            {
+                ["1"] = "developer",
+                ["2"] = "tester"
+            };
+        }
+
+        public ConcurrentDictionary<string, string> Users { get; set; }
+
+        public ConcurrentStack<Message> AllMessages { get; }
+
+        public Message AddMessage(ReceivedMessage message)
+        {
+            if (!Users.TryGetValue(message.FromId, out var displayName))
+            {
+                displayName = "(unknown)";
+            }
+
+            return AddMessage(new Message
+            {
+                Content = message.Content,
+                SentAt = message.SentAt,
+                From = new MessageFrom
+                {
+                    DisplayName = displayName,
+                    Id = message.FromId
+                }
+            });
+        }
+
+        public Message AddMessage(Message message)
+        {
+            AllMessages.Push(message);
+            _messageStream.OnNext(message);
+            return message;
+		}
+
+		public MessageFrom Join(string userId)
+		{
+			if (!Users.TryGetValue(userId, out var displayName))
+			{
+				displayName = "(unknown)";
+			}
+
+			var joinedUser = new MessageFrom {
+				Id = userId,
+				DisplayName = displayName
+			};
+
+			_userJoined.OnNext(joinedUser);
+			return joinedUser;
+		}
+
+		public IObservable<Message> Messages(string user)
+        {
+            return _messageStream
+                .Select(message =>
+                {
+                    message.Sub = user;
+                    return message;
+                })
+                .AsObservable();
+        }
+
+        public void AddError(Exception exception)
+        {
+            _messageStream.OnError(exception);
+        }
+
+		public IObservable<MessageFrom> UserJoined()
+		{
+			return _userJoined.AsObservable();
+		}
+	}
+
+	public class User
+	{
+		public string Id { get; set; }
+		public string Name { get; set; }
+	}
+}
diff --git a/tests/IntegrationTestServer/ChatSchema/Message.cs b/tests/IntegrationTestServer/ChatSchema/Message.cs
new file mode 100644
index 00000000..f3b90cb1
--- /dev/null
+++ b/tests/IntegrationTestServer/ChatSchema/Message.cs
@@ -0,0 +1,15 @@
+using System;
+
+namespace IntegrationTestServer.ChatSchema
+{
+    public class Message
+    {
+        public MessageFrom From { get; set; }
+
+        public string Sub { get; set; }
+
+        public string Content { get; set; }
+
+        public DateTime SentAt { get; set; }
+    }
+}
diff --git a/tests/IntegrationTestServer/ChatSchema/MessageFrom.cs b/tests/IntegrationTestServer/ChatSchema/MessageFrom.cs
new file mode 100644
index 00000000..5346cb6b
--- /dev/null
+++ b/tests/IntegrationTestServer/ChatSchema/MessageFrom.cs
@@ -0,0 +1,9 @@
+namespace IntegrationTestServer.ChatSchema
+{
+    public class MessageFrom
+    {
+        public string Id { get; set; }
+
+        public string DisplayName { get; set; }
+    }
+}
diff --git a/tests/IntegrationTestServer/ChatSchema/MessageFromType.cs b/tests/IntegrationTestServer/ChatSchema/MessageFromType.cs
new file mode 100644
index 00000000..98830475
--- /dev/null
+++ b/tests/IntegrationTestServer/ChatSchema/MessageFromType.cs
@@ -0,0 +1,13 @@
+using GraphQL.Types;
+
+namespace IntegrationTestServer.ChatSchema
+{
+    public class MessageFromType : ObjectGraphType<MessageFrom>
+    {
+        public MessageFromType()
+        {
+            Field(o => o.Id);
+            Field(o => o.DisplayName);
+        }
+    }
+}
diff --git a/tests/IntegrationTestServer/ChatSchema/MessageType.cs b/tests/IntegrationTestServer/ChatSchema/MessageType.cs
new file mode 100644
index 00000000..320f5b55
--- /dev/null
+++ b/tests/IntegrationTestServer/ChatSchema/MessageType.cs
@@ -0,0 +1,21 @@
+using GraphQL.Types;
+
+namespace IntegrationTestServer.ChatSchema
+{
+    public class MessageType : ObjectGraphType<Message>
+    {
+        public MessageType()
+        {
+            Field(o => o.Content);
+            Field(o => o.SentAt);
+            Field(o => o.Sub);
+            Field(o => o.From, false, typeof(MessageFromType)).Resolve(ResolveFrom);
+        }
+
+        private MessageFrom ResolveFrom(ResolveFieldContext<Message> context)
+        {
+            var message = context.Source;
+            return message.From;
+        }
+    }
+}
diff --git a/tests/IntegrationTestServer/ChatSchema/ReceivedMessage.cs b/tests/IntegrationTestServer/ChatSchema/ReceivedMessage.cs
new file mode 100644
index 00000000..0f1a409a
--- /dev/null
+++ b/tests/IntegrationTestServer/ChatSchema/ReceivedMessage.cs
@@ -0,0 +1,13 @@
+using System;
+
+namespace IntegrationTestServer.ChatSchema
+{
+    public class ReceivedMessage
+    {
+        public string FromId { get; set; }
+
+        public string Content { get; set; }
+
+        public DateTime SentAt { get; set; }
+    }
+}
diff --git a/tests/IntegrationTestServer/IntegrationTestServer.csproj b/tests/IntegrationTestServer/IntegrationTestServer.csproj
new file mode 100644
index 00000000..cfba48cd
--- /dev/null
+++ b/tests/IntegrationTestServer/IntegrationTestServer.csproj
@@ -0,0 +1,24 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp2.1</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <Folder Include="wwwroot\" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <PackageReference Include="GraphQL.Server.Transports.AspNetCore" Version="3.4.0" />
+    <PackageReference Include="GraphQL.Server.Transports.Subscriptions.Abstractions" Version="3.4.0" />
+    <PackageReference Include="GraphQL.Server.Transports.WebSockets" Version="3.4.0" />
+    <PackageReference Include="GraphQL.Server.Ui.GraphiQL" Version="3.4.0" />
+    <PackageReference Include="GraphQL.Server.Ui.Playground" Version="3.4.0" />
+    <PackageReference Include="GraphQL.Server.Ui.Voyager" Version="3.4.0" />
+    <PackageReference Include="Microsoft.AspNetCore.App" />
+    <PackageReference Include="Microsoft.AspNetCore.Razor.Design" Version="2.1.2" PrivateAssets="All" />
+    <PackageReference Include="System.Reactive" Version="4.1.2" />
+    <PackageReference Include="System.Reactive.Compatibility" Version="4.1.2" />
+  </ItemGroup>
+
+</Project>
diff --git a/tests/IntegrationTestServer/Program.cs b/tests/IntegrationTestServer/Program.cs
new file mode 100644
index 00000000..9b0392b7
--- /dev/null
+++ b/tests/IntegrationTestServer/Program.cs
@@ -0,0 +1,19 @@
+using Microsoft.AspNetCore;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace IntegrationTestServer
+{
+	public class Program
+	{
+		public static void Main(string[] args)
+		{
+			CreateWebHostBuilder(args).Build().Run();
+		}
+
+		public static IWebHostBuilder CreateWebHostBuilder(string[] args) =>
+			WebHost.CreateDefaultBuilder(args)
+				.UseStartup<Startup>()
+			    .ConfigureLogging((ctx, logging) => logging.SetMinimumLevel(LogLevel.Debug));
+	}
+}
diff --git a/tests/IntegrationTestServer/Properties/launchSettings.json b/tests/IntegrationTestServer/Properties/launchSettings.json
new file mode 100644
index 00000000..cd6aedd6
--- /dev/null
+++ b/tests/IntegrationTestServer/Properties/launchSettings.json
@@ -0,0 +1,28 @@
+{
+  "iisSettings": {
+    "windowsAuthentication": false,
+    "anonymousAuthentication": true,
+    "iisExpress": {
+      "applicationUrl": "http://localhost:55069/ui/graphiql/",
+      "sslPort": 0
+    }
+  },
+  "profiles": {
+    "IIS Express": {
+      "commandName": "IISExpress",
+      "launchBrowser": true,
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    },
+    "IntegrationTestServer": {
+      "commandName": "Project",
+      "launchBrowser": true,
+      "launchUrl": "ui/graphiql",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      },
+      "applicationUrl": "http://localhost:5005/"
+    }
+  }
+}
\ No newline at end of file
diff --git a/tests/IntegrationTestServer/Startup.cs b/tests/IntegrationTestServer/Startup.cs
new file mode 100644
index 00000000..166bf202
--- /dev/null
+++ b/tests/IntegrationTestServer/Startup.cs
@@ -0,0 +1,74 @@
+using GraphQL.Server;
+using GraphQL.Server.Ui.GraphiQL;
+using GraphQL.Server.Ui.Voyager;
+using GraphQL.Server.Ui.Playground;
+using IntegrationTestServer.ChatSchema;
+using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace IntegrationTestServer
+{
+	public class Startup
+	{
+		public Startup(IConfiguration configuration, IHostingEnvironment environment)
+		{
+			Configuration = configuration;
+			Environment = environment;
+		}
+
+		public IConfiguration Configuration { get; }
+		public IHostingEnvironment Environment { get; }
+
+		// This method gets called by the runtime. Use this method to add services to the container.
+		// For more information on how to configure your application, visit https://go.microsoft.com/fwlink/?LinkID=398940
+		public void ConfigureServices(IServiceCollection services)
+		{
+			services.AddSingleton<IChat, Chat>();
+			services.AddSingleton<ChatSchema.ChatSchema>();
+			services.AddSingleton<ChatQuery>();
+			services.AddSingleton<ChatMutation>();
+			services.AddSingleton<ChatSubscriptions>();
+			services.AddSingleton<MessageType>();
+			services.AddSingleton<MessageInputType>();
+
+			services.AddGraphQL(options =>
+				{
+					options.EnableMetrics = true;
+					options.ExposeExceptions = Environment.IsDevelopment();
+				})
+				.AddWebSockets();
+
+			services.AddMvc();
+		}
+
+		// This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
+		public void Configure(IApplicationBuilder app, IHostingEnvironment env)
+		{
+			if (env.IsDevelopment())
+			{
+				app.UseDeveloperExceptionPage();
+			}
+
+			app.UseWebSockets();
+			app.UseGraphQLWebSockets<ChatSchema.ChatSchema>("/graphql");
+			app.UseGraphQL<ChatSchema.ChatSchema>("/graphql");
+			app.UseGraphiQLServer(new GraphiQLOptions
+			{
+				GraphiQLPath = "/ui/graphiql",
+				GraphQLEndPoint = "/graphql"
+			});
+			app.UseGraphQLVoyager(new GraphQLVoyagerOptions()
+			{
+				GraphQLEndPoint = "/graphql",
+				Path = "/ui/voyager"
+			});
+			app.UseGraphQLPlayground(new GraphQLPlaygroundOptions
+			{
+				Path = "/ui/playground"
+			});
+			app.UseMvc();
+		}
+	}
+}