Rename to queue execution mode

LukeButters · LukeButters · commit 51b2da162472 · 2026-01-05T16:40:12.000+11:00
diff --git a/source/Halibut.Tests/RPCOverQueueExecutionModeFixture.cs b/source/Halibut.Tests/RPCOverQueueExecutionModeFixture.cs
@@ -21,11 +21,11 @@
 
 namespace Halibut.Tests
 {
-    public class LocalExecutionModeFixture : BaseTest
+    public class RPCOverQueueExecutionModeFixture : BaseTest
     {
         [RedisTest]
         [Test]
-        public async Task SimpleLocalExecutionExample()
+        public async Task SimpleRPCOverQueueExecutionExample()
         {
             var services = GetDelegateServiceFactory();
             var timeoutsAndLimits = new HalibutTimeoutsAndLimitsForTestsBuilder().Build();
@@ -55,7 +55,7 @@ public async Task SimpleLocalExecutionExample()
             using var workerCts = new CancellationTokenSource();
             var pollingTask = Task.Run(async () =>
             {
-                await worker.PollLocalAsync(new Uri("local://test-worker"), workerCts.Token);
+                await worker.PollForRPCOverQueueAsync(new Uri("local://test-worker"), workerCts.Token);
             }, workerCts.Token);
 
             // Client creates proxy to local://test-worker and makes request
diff --git a/source/Halibut/HalibutRuntime.cs b/source/Halibut/HalibutRuntime.cs
@@ -21,6 +21,7 @@ namespace Halibut
 {
     public class HalibutRuntime : IHalibutRuntime
     {
+        public const string QueueEndpointScheme = "local";
         public static readonly string DefaultFriendlyHtmlPageContent = "<html><body><p>Hello!</p></body></html>";
         readonly ConcurrentDictionary<Uri, IPendingRequestQueue> queues = new();
         readonly IPendingRequestQueueFactory queueFactory;
@@ -199,17 +200,17 @@ public void Poll(Uri subscription, ServiceEndPoint endPoint, CancellationToken c
             pollingClients.Add(new PollingClient(subscription, client, HandleIncomingRequestAsync, log, cancellationToken, pollingReconnectRetryPolicy));
         }
 
-        public async Task PollLocalAsync(Uri localEndpoint, CancellationToken cancellationToken)
+        public async Task PollForRPCOverQueueAsync(Uri queueOnlyEndpoint, CancellationToken cancellationToken)
         {
-            if (localEndpoint.Scheme.ToLowerInvariant() != "local")
+            if (queueOnlyEndpoint.Scheme.ToLowerInvariant() != QueueEndpointScheme)
             {
-                throw new ArgumentException($"Only 'local://' endpoints are supported. Provided: {localEndpoint.Scheme}://", nameof(localEndpoint));
+                throw new ArgumentException($"Only 'queue://' endpoints are supported. Provided: {queueOnlyEndpoint.Scheme}://", nameof(queueOnlyEndpoint));
             }
 
-            var queue = GetQueue(localEndpoint);
-            var log = logs.ForEndpoint(localEndpoint);
+            var queue = GetQueue(queueOnlyEndpoint);
+            var log = logs.ForEndpoint(queueOnlyEndpoint);
 
-            log.Write(EventType.MessageExchange, $"Starting local polling for endpoint: {localEndpoint}");
+            log.Write(EventType.MessageExchange, $"Starting queue polling for endpoint: {queueOnlyEndpoint}");
 
             while (!cancellationToken.IsCancellationRequested)
             {
@@ -226,7 +227,7 @@ public async Task PollLocalAsync(Uri localEndpoint, CancellationToken cancellati
                         }
                         catch (Exception ex)
                         {
-                            log.WriteException(EventType.Error, $"Error executing local request for {request.RequestMessage.ServiceName}.{request.RequestMessage.MethodName}", ex);
+                            log.WriteException(EventType.Error, $"Error executing queue request for {request.RequestMessage.ServiceName}.{request.RequestMessage.MethodName}", ex);
                             response = ResponseMessage.FromException(request.RequestMessage, ex);
                         }
 
@@ -235,17 +236,17 @@ public async Task PollLocalAsync(Uri localEndpoint, CancellationToken cancellati
                 }
                 catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
                 {
-                    log.Write(EventType.MessageExchange, $"Local polling cancelled for endpoint: {localEndpoint}");
+                    log.Write(EventType.MessageExchange, $"Queue polling cancelled for endpoint: {queueOnlyEndpoint}");
                     break;
                 }
                 catch (Exception ex)
                 {
-                    log.WriteException(EventType.Error, $"Error in local polling loop for endpoint: {localEndpoint}", ex);
+                    log.WriteException(EventType.Error, $"Error in queue polling loop for endpoint: {queueOnlyEndpoint}", ex);
                     await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
                 }
             }
 
-            log.Write(EventType.MessageExchange, $"Local polling stopped for endpoint: {localEndpoint}");
+            log.Write(EventType.MessageExchange, $"Queue polling stopped for endpoint: {queueOnlyEndpoint}");
         }
 
         public async Task<ServiceEndPoint> DiscoverAsync(Uri uri, CancellationToken cancellationToken)
@@ -291,9 +292,7 @@ async Task<ResponseMessage> SendOutgoingRequestAsync(RequestMessage request, Met
                         response = await SendOutgoingHttpsRequestAsync(request, cancellationToken).ConfigureAwait(false);
                         break;
                     case "poll":
-                        response = await SendOutgoingPollingRequestAsync(request, cancellationToken).ConfigureAwait(false);
-                        break;
-                    case "local":
+                    case QueueEndpointScheme:
                         response = await SendOutgoingPollingRequestAsync(request, cancellationToken).ConfigureAwait(false);
                         break;
                     default: throw new ArgumentException("Unknown endpoint type: " + endPoint.BaseUri.Scheme);
diff --git a/source/Halibut/IHalibutRuntime.cs b/source/Halibut/IHalibutRuntime.cs
@@ -17,7 +17,7 @@ public interface IHalibutRuntime : IAsyncDisposable, IDisposable
         int Listen(IPEndPoint endpoint);
         void ListenWebSocket(string endpoint);
         void Poll(Uri subscription, ServiceEndPoint endPoint, CancellationToken cancellationToken);
-        Task PollLocalAsync(Uri localEndpoint, CancellationToken cancellationToken);
+        Task PollForRPCOverQueueAsync(Uri queueOnlyEndpoint, CancellationToken cancellationToken);
 
         Task<ServiceEndPoint> DiscoverAsync(Uri uri, CancellationToken cancellationToken);
         Task<ServiceEndPoint> DiscoverAsync(ServiceEndPoint endpoint, CancellationToken cancellationToken);

Original file line number	Diff line number	Diff line change
`@@ -21,11 +21,11 @@`
`21`	`21`
`22`	`22`	`namespace Halibut.Tests`
`23`	`23`	`{`
`24`		`- public class LocalExecutionModeFixture : BaseTest`
	`24`	`+ public class RPCOverQueueExecutionModeFixture : BaseTest`
`25`	`25`	`{`
`26`	`26`	`[RedisTest]`
`27`	`27`	`[Test]`
`28`		`- public async Task SimpleLocalExecutionExample()`
	`28`	`+ public async Task SimpleRPCOverQueueExecutionExample()`
`29`	`29`	`{`
`30`	`30`	`var services = GetDelegateServiceFactory();`
`31`	`31`	`var timeoutsAndLimits = new HalibutTimeoutsAndLimitsForTestsBuilder().Build();`
`@@ -55,7 +55,7 @@ public async Task SimpleLocalExecutionExample()`
`55`	`55`	`using var workerCts = new CancellationTokenSource();`
`56`	`56`	`var pollingTask = Task.Run(async () =>`
`57`	`57`	`{`
`58`		`- await worker.PollLocalAsync(new Uri("local://test-worker"), workerCts.Token);`
	`58`	`+ await worker.PollForRPCOverQueueAsync(new Uri("local://test-worker"), workerCts.Token);`
`59`	`59`	`}, workerCts.Token);`
`60`	`60`
`61`	`61`	`// Client creates proxy to local://test-worker and makes request`
Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,7 @@ namespace Halibut`
`21`	`21`	`{`
`22`	`22`	`public class HalibutRuntime : IHalibutRuntime`
`23`	`23`	`{`
	`24`	`+ public const string QueueEndpointScheme = "local";`
`24`	`25`	`public static readonly string DefaultFriendlyHtmlPageContent = "<html><body><p>Hello!</p></body></html>";`
`25`	`26`	`readonly ConcurrentDictionary<Uri, IPendingRequestQueue> queues = new();`
`26`	`27`	`readonly IPendingRequestQueueFactory queueFactory;`
`@@ -199,17 +200,17 @@ public void Poll(Uri subscription, ServiceEndPoint endPoint, CancellationToken c`
`199`	`200`	`pollingClients.Add(new PollingClient(subscription, client, HandleIncomingRequestAsync, log, cancellationToken, pollingReconnectRetryPolicy));`
`200`	`201`	`}`
`201`	`202`
`202`		`- public async Task PollLocalAsync(Uri localEndpoint, CancellationToken cancellationToken)`
	`203`	`+ public async Task PollForRPCOverQueueAsync(Uri queueOnlyEndpoint, CancellationToken cancellationToken)`
`203`	`204`	`{`
`204`		`- if (localEndpoint.Scheme.ToLowerInvariant() != "local")`
	`205`	`+ if (queueOnlyEndpoint.Scheme.ToLowerInvariant() != QueueEndpointScheme)`
`205`	`206`	`{`
`206`		`- throw new ArgumentException($"Only 'local://' endpoints are supported. Provided: {localEndpoint.Scheme}://", nameof(localEndpoint));`
	`207`	`+ throw new ArgumentException($"Only 'queue://' endpoints are supported. Provided: {queueOnlyEndpoint.Scheme}://", nameof(queueOnlyEndpoint));`
`207`	`208`	`}`
`208`	`209`
`209`		`- var queue = GetQueue(localEndpoint);`
`210`		`- var log = logs.ForEndpoint(localEndpoint);`
	`210`	`+ var queue = GetQueue(queueOnlyEndpoint);`
	`211`	`+ var log = logs.ForEndpoint(queueOnlyEndpoint);`
`211`	`212`
`212`		`- log.Write(EventType.MessageExchange, $"Starting local polling for endpoint: {localEndpoint}");`
	`213`	`+ log.Write(EventType.MessageExchange, $"Starting queue polling for endpoint: {queueOnlyEndpoint}");`
`213`	`214`
`214`	`215`	`while (!cancellationToken.IsCancellationRequested)`
`215`	`216`	`{`
`@@ -226,7 +227,7 @@ public async Task PollLocalAsync(Uri localEndpoint, CancellationToken cancellati`
`226`	`227`	`}`
`227`	`228`	`catch (Exception ex)`
`228`	`229`	`{`
`229`		`- log.WriteException(EventType.Error, $"Error executing local request for {request.RequestMessage.ServiceName}.{request.RequestMessage.MethodName}", ex);`
	`230`	`+ log.WriteException(EventType.Error, $"Error executing queue request for {request.RequestMessage.ServiceName}.{request.RequestMessage.MethodName}", ex);`
`230`	`231`	`response = ResponseMessage.FromException(request.RequestMessage, ex);`
`231`	`232`	`}`
`232`	`233`
`@@ -235,17 +236,17 @@ public async Task PollLocalAsync(Uri localEndpoint, CancellationToken cancellati`
`235`	`236`	`}`
`236`	`237`	`catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)`
`237`	`238`	`{`
`238`		`- log.Write(EventType.MessageExchange, $"Local polling cancelled for endpoint: {localEndpoint}");`
	`239`	`+ log.Write(EventType.MessageExchange, $"Queue polling cancelled for endpoint: {queueOnlyEndpoint}");`
`239`	`240`	`break;`
`240`	`241`	`}`
`241`	`242`	`catch (Exception ex)`
`242`	`243`	`{`
`243`		`- log.WriteException(EventType.Error, $"Error in local polling loop for endpoint: {localEndpoint}", ex);`
	`244`	`+ log.WriteException(EventType.Error, $"Error in queue polling loop for endpoint: {queueOnlyEndpoint}", ex);`
`244`	`245`	`await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);`
`245`	`246`	`}`
`246`	`247`	`}`
`247`	`248`
`248`		`- log.Write(EventType.MessageExchange, $"Local polling stopped for endpoint: {localEndpoint}");`
	`249`	`+ log.Write(EventType.MessageExchange, $"Queue polling stopped for endpoint: {queueOnlyEndpoint}");`
`249`	`250`	`}`
`250`	`251`
`251`	`252`	`public async Task<ServiceEndPoint> DiscoverAsync(Uri uri, CancellationToken cancellationToken)`
`@@ -291,9 +292,7 @@ async Task<ResponseMessage> SendOutgoingRequestAsync(RequestMessage request, Met`
`291`	`292`	`response = await SendOutgoingHttpsRequestAsync(request, cancellationToken).ConfigureAwait(false);`
`292`	`293`	`break;`
`293`	`294`	`case "poll":`
`294`		`- response = await SendOutgoingPollingRequestAsync(request, cancellationToken).ConfigureAwait(false);`
`295`		`- break;`
`296`		`- case "local":`
	`295`	`+ case QueueEndpointScheme:`
`297`	`296`	`response = await SendOutgoingPollingRequestAsync(request, cancellationToken).ConfigureAwait(false);`
`298`	`297`	`break;`
`299`	`298`	`default: throw new ArgumentException("Unknown endpoint type: " + endPoint.BaseUri.Scheme);`