Async tweaks

Author: pawelpabich

source/Halibut.Tests/Support/PendingRequestQueueFactories/CancelWhenRequestDequeuedPendingRequestQueueFactory.cs

@@ -29,7 +29,7 @@ public IPendingRequestQueue CreateQueue(Uri endpoint)
             return new Decorator(inner.CreateQueue(endpoint), cancellationTokenSource, shouldCancelOnDequeue, onResponseApplied);
         }
 
-        public Task<IPendingRequestQueue> CreateQueueAsync(Uri endpoint)
+        public Task<IPendingRequestQueue> CreateQueueAsync(Uri endpoint, CancellationToken cancellationToken)
         {
             return Task.FromResult(CreateQueue(endpoint));
         }

source/Halibut.Tests/Support/PendingRequestQueueFactories/CancelWhenRequestQueuedPendingRequestQueueFactory.cs

@@ -28,7 +28,7 @@ public IPendingRequestQueue CreateQueue(Uri endpoint)
             return new Decorator(inner.CreateQueue(endpoint), cancellationTokenSources);
         }
 
-        public Task<IPendingRequestQueue> CreateQueueAsync(Uri endpoint)
+        public Task<IPendingRequestQueue> CreateQueueAsync(Uri endpoint, CancellationToken cancellationToken)
         {
             return Task.FromResult(CreateQueue(endpoint));
         }

source/Halibut.Tests/Util/FuncPendingRequestQueueFactory.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using Halibut.ServiceModel;
 
@@ -18,7 +19,7 @@ public IPendingRequestQueue CreateQueue(Uri endpoint)
             return createQueue(endpoint);
         }
 
-        public Task<IPendingRequestQueue> CreateQueueAsync(Uri endpoint)
+        public Task<IPendingRequestQueue> CreateQueueAsync(Uri endpoint, CancellationToken cancellationToken)
         {
             return Task.FromResult(createQueue(endpoint));
         }

source/Halibut/DataStream.cs

@@ -10,7 +10,7 @@ namespace Halibut
 {
     public class DataStream : IEquatable<DataStream>, IDataStreamInternal
     {
-        readonly Func<Stream, CancellationToken, Task> writerAsync;
+        protected readonly Func<Stream, CancellationToken, Task> writerAsync;
         IDataStreamReceiver? receiver;
 
         [JsonConstructor]

source/Halibut/HalibutRuntime.cs

@@ -90,7 +90,7 @@ internal HalibutRuntime(
 
         IPendingRequestQueue GetQueue(Uri target)
         {
-            return queues.GetOrAdd(target, u => queueFactory.CreateQueueAsync(target).Result);
+            return queues.GetOrAdd(target, u => queueFactory.CreateQueue(target));
         }
 
         public int Listen()

source/Halibut/HalibutRuntimeBuilder.cs

@@ -126,7 +126,7 @@ public HalibutRuntimeBuilder WithRpcObserver(IRpcObserver rpcObserver)
             return this;
         }
 
-        public HalibutRuntimeBuilder WithIdentityObserver(ISubscribersObserver subscribersObserver)
+        public HalibutRuntimeBuilder WithSubscribersObserver(ISubscribersObserver subscribersObserver)
         {
             this.identityObserver = subscribersObserver;
             return this;

source/Halibut/ServiceModel/IPendingRequestQueueFactory.cs

@@ -1,11 +1,12 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace Halibut.ServiceModel
 {
     public interface IPendingRequestQueueFactory
     {
         IPendingRequestQueue CreateQueue(Uri endpoint);
-        Task<IPendingRequestQueue> CreateQueueAsync(Uri endpoint);
+        Task<IPendingRequestQueue> CreateQueueAsync(Uri endpoint, CancellationToken cancellationToken);
     }
 }

source/Halibut/ServiceModel/PendingRequestQueueFactoryAsync.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using Halibut.Diagnostics;
 
@@ -20,7 +21,7 @@ public IPendingRequestQueue CreateQueue(Uri endpoint)
             return new PendingRequestQueueAsync(halibutTimeoutsAndLimits, logFactory.ForEndpoint(endpoint));
         }
 
-        public Task<IPendingRequestQueue> CreateQueueAsync(Uri endpoint)
+        public Task<IPendingRequestQueue> CreateQueueAsync(Uri endpoint, CancellationToken cancellationToken)
         {
             return Task.FromResult(CreateQueue(endpoint));
         }