Very basic subscription tests and client support

src/Tgstation.Server.Client.GraphQL/.graphqlrc.json

@@ -13,7 +13,7 @@
       "transportProfiles": [
         {
           "default": "Http",
-          "subscription": "WebSocket"
+          "subscription": "Http"
         }
       ]
     }

src/Tgstation.Server.Client.GraphQL/GQL/Subscriptions/SessionInvalidation.graphql

@@ -0,0 +1,3 @@
+subscription SessionInvalidation {
+  sessionInvalidated
+}

src/Tgstation.Server.Client.GraphQL/GraphQLServerClient.cs

@@ -124,19 +124,47 @@ namespace Tgstation.Server.Client.GraphQL
 		public virtual ValueTask DisposeAsync() => serviceProvider.DisposeAsync();
 
 		/// <inheritdoc />
-		public ValueTask<IOperationResult<TResultData>> RunOperationAsync<TResultData>(Func<IGraphQLClient, ValueTask<IOperationResult<TResultData>>> queryExector, CancellationToken cancellationToken)
+		public ValueTask<IOperationResult<TResultData>> RunOperationAsync<TResultData>(Func<IGraphQLClient, ValueTask<IOperationResult<TResultData>>> operationExecutor, CancellationToken cancellationToken)
 			where TResultData : class
 		{
-			ArgumentNullException.ThrowIfNull(queryExector);
-			return WrapAuthentication(queryExector, cancellationToken);
+			ArgumentNullException.ThrowIfNull(operationExecutor);
+			return WrapAuthentication(operationExecutor, cancellationToken);
 		}
 
 		/// <inheritdoc />
-		public ValueTask<IOperationResult<TResultData>> RunOperation<TResultData>(Func<IGraphQLClient, Task<IOperationResult<TResultData>>> queryExector, CancellationToken cancellationToken)
+		public ValueTask<IOperationResult<TResultData>> RunOperation<TResultData>(Func<IGraphQLClient, Task<IOperationResult<TResultData>>> operationExecutor, CancellationToken cancellationToken)
 			where TResultData : class
 		{
-			ArgumentNullException.ThrowIfNull(queryExector);
-			return WrapAuthentication(async localClient => await queryExector(localClient), cancellationToken);
+			ArgumentNullException.ThrowIfNull(operationExecutor);
+			return WrapAuthentication(async localClient => await operationExecutor(localClient), cancellationToken);
+		}
+
+		/// <inheritdoc />
+		public async ValueTask<IDisposable> Subscribe<TResultData>(Func<IGraphQLClient, IObservable<IOperationResult<TResultData>>> operationExecutor, IObserver<IOperationResult<TResultData>> observer, CancellationToken cancellationToken)
+			where TResultData : class
+		{
+			ArgumentNullException.ThrowIfNull(operationExecutor);
+			ArgumentNullException.ThrowIfNull(observer);
+
+			var observable = operationExecutor(graphQLClient);
+
+			if (Authenticated)
+			{
+				var tuple = await bearerCredentialsTask.ConfigureAwait(false);
+				if (!tuple.HasValue)
+					ThrowOtherCallerFailedAuthException();
+
+				var (currentAuthHeader, expires) = tuple.Value;
+				if (expires <= DateTimeOffset.UtcNow)
+					currentAuthHeader = await Reauthenticate(currentAuthHeader, cancellationToken).ConfigureAwait(false);
+
+				setAuthenticationHeader(currentAuthHeader);
+			}
+
+			// maybe make this handle reauthentication one day
+			// but would need to check if lost auth results in complete events being sent
+			// if so, it can't be done
+			return observable.Subscribe(observer);
 		}
 
 		/// <summary>
@@ -167,59 +195,6 @@ namespace Tgstation.Server.Client.GraphQL
 			if (!tuple.HasValue)
 				ThrowOtherCallerFailedAuthException();
 
-			async ValueTask<AuthenticationHeaderValue> Reauthenticate(AuthenticationHeaderValue currentToken, CancellationToken cancellationToken)
-			{
-				if (!CanReauthenticate)
-					throw new AuthenticationException("Authentication expired or invalid and cannot re-authenticate.");
-
-				TaskCompletionSource<(AuthenticationHeaderValue Header, DateTime Exp)?>? tcs = null;
-				do
-				{
-					var bearerCredentialsTaskLocal = bearerCredentialsTask;
-					if (!bearerCredentialsTaskLocal!.IsCompleted)
-					{
-						var currentTuple = await bearerCredentialsTaskLocal.ConfigureAwait(false);
-						if (!currentTuple.HasValue)
-							ThrowOtherCallerFailedAuthException();
-
-						return currentTuple.Value.Header;
-					}
-
-					lock (bearerCredentialsHeaderTaskLock!)
-					{
-						if (bearerCredentialsTask == bearerCredentialsTaskLocal)
-						{
-							var result = bearerCredentialsTaskLocal.Result;
-							if (result?.Header != currentToken)
-							{
-								if (!result.HasValue)
-									ThrowOtherCallerFailedAuthException();
-
-								return result.Value.Header;
-							}
-
-							tcs = new TaskCompletionSource<(AuthenticationHeaderValue, DateTime)?>();
-							bearerCredentialsTask = tcs.Task;
-						}
-					}
-				}
-				while (tcs == null);
-
-				setAuthenticationHeader!(basicCredentialsHeader!);
-				var loginResult = await graphQLClient.Login.ExecuteAsync(cancellationToken).ConfigureAwait(false);
-				try
-				{
-					var tuple = await CreateCredentialsTuple(loginResult).ConfigureAwait(false);
-					tcs.SetResult(tuple);
-					return tuple.Header;
-				}
-				catch (AuthenticationException)
-				{
-					tcs.SetResult(null);
-					throw;
-				}
-			}
-
 			var (currentAuthHeader, expires) = tuple.Value;
 			if (expires <= DateTimeOffset.UtcNow)
 				currentAuthHeader = await Reauthenticate(currentAuthHeader, cancellationToken).ConfigureAwait(false);
@@ -238,6 +213,65 @@ namespace Tgstation.Server.Client.GraphQL
 			return operationResult;
 		}
 
+		/// <summary>
+		/// Attempt to reauthenticate.
+		/// </summary>
+		/// <param name="currentToken">The current <see cref="AuthenticationHeaderValue"/> for the bearer token.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask{TResult}"/> resulting in the updated <see cref="AuthenticationHeaderValue"/> to use.</returns>
+		async ValueTask<AuthenticationHeaderValue> Reauthenticate(AuthenticationHeaderValue currentToken, CancellationToken cancellationToken)
+		{
+			if (!CanReauthenticate)
+				throw new AuthenticationException("Authentication expired or invalid and cannot re-authenticate.");
+
+			TaskCompletionSource<(AuthenticationHeaderValue Header, DateTime Exp)?>? tcs = null;
+			do
+			{
+				var bearerCredentialsTaskLocal = bearerCredentialsTask;
+				if (!bearerCredentialsTaskLocal!.IsCompleted)
+				{
+					var currentTuple = await bearerCredentialsTaskLocal.ConfigureAwait(false);
+					if (!currentTuple.HasValue)
+						ThrowOtherCallerFailedAuthException();
+
+					return currentTuple.Value.Header;
+				}
+
+				lock (bearerCredentialsHeaderTaskLock!)
+				{
+					if (bearerCredentialsTask == bearerCredentialsTaskLocal)
+					{
+						var result = bearerCredentialsTaskLocal.Result;
+						if (result?.Header != currentToken)
+						{
+							if (!result.HasValue)
+								ThrowOtherCallerFailedAuthException();
+
+							return result.Value.Header;
+						}
+
+						tcs = new TaskCompletionSource<(AuthenticationHeaderValue, DateTime)?>();
+						bearerCredentialsTask = tcs.Task;
+					}
+				}
+			}
+			while (tcs == null);
+
+			setAuthenticationHeader!(basicCredentialsHeader!);
+			var loginResult = await graphQLClient.Login.ExecuteAsync(cancellationToken).ConfigureAwait(false);
+			try
+			{
+				var tuple = await CreateCredentialsTuple(loginResult).ConfigureAwait(false);
+				tcs.SetResult(tuple);
+				return tuple.Header;
+			}
+			catch (AuthenticationException)
+			{
+				tcs.SetResult(null);
+				throw;
+			}
+		}
+
 		/// <summary>
 		/// Attempt to create the <see cref="ValueTuple{T1, T2}"/> for <see cref="bearerCredentialsTask"/>.
 		/// </summary>

src/Tgstation.Server.Client.GraphQL/IGraphQLServerClient.cs

@@ -32,5 +32,16 @@ namespace Tgstation.Server.Client.GraphQL
 		/// <exception cref="AuthenticationException">Thrown when automatic reauthentication fails.</exception>
 		ValueTask<IOperationResult<TResultData>> RunOperation<TResultData>(Func<IGraphQLClient, Task<IOperationResult<TResultData>>> operationExecutor, CancellationToken cancellationToken)
 			where TResultData : class;
+
+		/// <summary>
+		/// Subcribes to the GraphQL subscription indicated by <paramref name="operationExecutor"/>.
+		/// </summary>
+		/// <typeparam name="TResultData">The <see cref="Type"/> of the <see cref="IOperationResult{TResultData}"/>'s <see cref="IOperationResult{TResultData}.Data"/>.</typeparam>
+		/// <param name="operationExecutor">A <see cref="Func{T, TResult}"/> which initiates a single subscription on a given <see cref="IGraphQLClient"/> and returns a <see cref="ValueTask{TResult}"/> resulting in the <typeparamref name="TResultData"/> <see cref="IOperationResult{TResultData}"/> <see cref="IObservable{T}"/>.</param>
+		/// <param name="observer">The <see cref="IObserver{T}"/> for <typeparamref name="TResultData"/> <see cref="IOperationResult{TResultData}"/>s.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask{TResult}"/> resulting in the <see cref="IDisposable"/> representing the lifetime of the subscription.</returns>
+		ValueTask<IDisposable> Subscribe<TResultData>(Func<IGraphQLClient, IObservable<IOperationResult<TResultData>>> operationExecutor, IObserver<IOperationResult<TResultData>> observer, CancellationToken cancellationToken)
+			where TResultData : class;
 	}
 }

tests/Tgstation.Server.Tests/Live/HoldLastObserver.cs

@@ -0,0 +1,34 @@
+using System;
+
+namespace Tgstation.Server.Tests.Live
+{
+	sealed class HoldLastObserver<T> : IObserver<T>
+	{
+		public bool Completed { get; private set; }
+
+		public Exception LastError { get; private set; }
+
+		public T LastValue { get; private set; }
+
+		public ulong ErrorCount { get; private set; }
+
+		public ulong ResultCount { get; private set; }
+
+		public void OnCompleted()
+		{
+			Completed = true;
+		}
+
+		public void OnError(Exception error)
+		{
+			++ErrorCount;
+			LastError = error;
+		}
+
+		public void OnNext(T value)
+		{
+			++ResultCount;
+			LastValue = value;
+		}
+	}
+}

tests/Tgstation.Server.Tests/Live/TestLiveServer.cs

@@ -1395,6 +1395,19 @@ namespace Tgstation.Server.Tests.Live
 							return result;
 						},
 						cancellationToken);
+
+					var testObserver = new HoldLastObserver<IOperationResult<ISessionInvalidationResult>>();
+					using var subscription = await unauthenticatedGraphQLClient.Subscribe(
+						gql => gql.SessionInvalidation.Watch(),
+						testObserver,
+						cancellationToken);
+
+					await Task.Delay(1000, cancellationToken);
+
+					Assert.AreEqual(0U, testObserver.ErrorCount);
+					Assert.AreEqual(1U, testObserver.ResultCount);
+					Assert.IsTrue(testObserver.LastValue.IsAuthenticationError());
+					Assert.IsTrue(testObserver.Completed);
 				}
 
 				async ValueTask<MultiServerClient> CreateUserWithNoInstancePerms()
@@ -1416,6 +1429,8 @@ namespace Tgstation.Server.Tests.Live
 					return await CreateClient(server.RootUrl, createRequest.Name, createRequest.Password, false, cancellationToken);
 				}
 
+				var restartObserver = new HoldLastObserver<IOperationResult<ISessionInvalidationResult>>();
+				IDisposable restartSubscription;
 				var jobsHubTest = new JobsHubTests(firstAdminMultiClient, await CreateUserWithNoInstancePerms());
 				Task jobsHubTestTask;
 				{
@@ -1579,12 +1594,45 @@ namespace Tgstation.Server.Tests.Live
 					initialStaged = dd.StagedCompileJob.Id.Value;
 					initialSessionId = dd.SessionId.Value;
 
-					jobsHubTest.ExpectShutdown();
-					await firstAdminRestClient.Administration.Restart(cancellationToken);
+					// force a session refresh if necessary
+					await firstAdminMultiClient.GraphQLClient.RunQueryEnsureNoErrors(
+						gql => gql.ReadCurrentUser.ExecuteAsync(cancellationToken),
+						cancellationToken);
+
+					restartSubscription = await firstAdminMultiClient.GraphQLClient.Subscribe(
+						gql => gql.SessionInvalidation.Watch(),
+						restartObserver,
+						cancellationToken);
+
+					try
+					{
+						await Task.Delay(1000, cancellationToken);
+
+						jobsHubTest.ExpectShutdown();
+						await firstAdminRestClient.Administration.Restart(cancellationToken);
+					}
+					catch
+					{
+						restartSubscription.Dispose();
+						throw;
+					}
 				}
 
-				await Task.WhenAny(serverTask, Task.Delay(TimeSpan.FromMinutes(1), cancellationToken));
-				Assert.IsTrue(serverTask.IsCompleted);
+				try
+				{
+					await Task.WhenAny(serverTask, Task.Delay(TimeSpan.FromMinutes(1), cancellationToken));
+					Assert.IsTrue(serverTask.IsCompleted);
+
+					Assert.AreEqual(0U, restartObserver.ErrorCount);
+					Assert.AreEqual(1U, restartObserver.ResultCount);
+					restartObserver.LastValue.EnsureNoErrors();
+					Assert.IsTrue(restartObserver.Completed);
+					Assert.AreEqual(SessionInvalidationReason.ServerShutdown, restartObserver.LastValue.Data.SessionInvalidated);
+				}
+				finally
+				{
+					restartSubscription.Dispose();
+				}
 
 				// test the reattach message queueing
 				// for the code coverage really...