allow fifo task execution

GrpcDotNetNamedPipes/Internal/ServerConnectionContext.cs

@@ -21,9 +21,10 @@ internal class ServerConnectionContext : TransportMessageHandler, IDisposable
     private readonly ConnectionLogger _logger;
     private readonly Dictionary<string, Func<ServerConnectionContext, Task>> _methodHandlers;
     private readonly PayloadQueue _payloadQueue;
-
+    private readonly TaskFactory _taskFactory;
+
     public ServerConnectionContext(NamedPipeServerStream pipeStream, ConnectionLogger logger,
-        Dictionary<string, Func<ServerConnectionContext, Task>> methodHandlers)
+        Dictionary<string, Func<ServerConnectionContext, Task>> methodHandlers, TaskFactory taskFactory)
     {
         CallContext = new NamedPipeCallContext(this);
         PipeStream = pipeStream;
@@ -32,6 +33,7 @@ public ServerConnectionContext(NamedPipeServerStream pipeStream, ConnectionLogge
         _methodHandlers = methodHandlers;
         _payloadQueue = new PayloadQueue();
         CancellationTokenSource = new CancellationTokenSource();
+        _taskFactory = taskFactory;
     }
 
     public NamedPipeServerStream PipeStream { get; }
@@ -58,11 +60,13 @@ public IServerStreamWriter<TResponse> CreateResponseStream<TResponse>(Marshaller
         return new ResponseStreamWriterImpl<TResponse>(Transport, CancellationToken.None, responseMarshaller,
             () => IsCompleted);
     }
-
+    
     public override void HandleRequestInit(string methodFullName, DateTime? deadline)
     {
         Deadline = new Deadline(deadline);
-        Task.Run(async () => await _methodHandlers[methodFullName](this).ConfigureAwait(false));
+        // Note the use of .ConfigureAwait(false) here...
+        // https://blog.stephencleary.com/2012/07/dont-block-on-async-code.html
+        _taskFactory.StartNew(async () => await _methodHandlers[methodFullName](this).ConfigureAwait(false)); 
     }
 
     public override void HandleHeaders(Metadata headers) => RequestHeaders = headers;

GrpcDotNetNamedPipes/Internal/ServerStreamPool.cs

@@ -161,7 +161,8 @@ private void RunHandleConnection(NamedPipeServerStream pipeServer)
             try
             {
                 await _handleConnection(pipeServer);
-                pipeServer.Disconnect();
+                if (pipeServer.IsConnected)
+                    pipeServer.Disconnect();
             }
             catch (Exception error)
             {

GrpcDotNetNamedPipes/NamedPipeServer.cs

@@ -19,6 +19,7 @@ namespace GrpcDotNetNamedPipes;
 public class NamedPipeServer : IDisposable
 {
     private readonly ServerStreamPool _pool;
+    private readonly TaskFactory _taskFactory;
     private readonly Action<string> _log;
     private readonly Dictionary<string, Func<ServerConnectionContext, Task>> _methodHandlers = new();
 
@@ -32,11 +33,12 @@ public NamedPipeServer(string pipeName, NamedPipeServerOptions options)
     {
     }
 
-    internal NamedPipeServer(string pipeName, NamedPipeServerOptions options, Action<string> log)
+    public NamedPipeServer(string pipeName, NamedPipeServerOptions options, Action<string> log)
     {
         _pool = new ServerStreamPool(pipeName, options, HandleConnection, InvokeError);
         _log = log;
         ServiceBinder = new ServiceBinderImpl(this);
+        _taskFactory = options.TaskFactory ?? new TaskFactory();
     }
 
     public ServiceBinderBase ServiceBinder { get; }
@@ -66,7 +68,7 @@ public void Dispose()
     private async Task HandleConnection(NamedPipeServerStream pipeStream)
     {
         var logger = ConnectionLogger.Server(_log);
-        var ctx = new ServerConnectionContext(pipeStream, logger, _methodHandlers);
+        var ctx = new ServerConnectionContext(pipeStream, logger, _methodHandlers, _taskFactory);
         await Task.Run(new PipeReader(pipeStream, ctx, logger, ctx.Dispose, InvokeError).ReadLoop);
     }
 

GrpcDotNetNamedPipes/NamedPipeServerOptions.cs

@@ -31,4 +31,11 @@ public class NamedPipeServerOptions
     /// </summary>
     public PipeSecurity PipeSecurity { get; set; }
 #endif
+
+    /// <summary>
+    /// Gets or sets a Custom Task Factory to control how tasks are serviced. 
+    /// For example, causing threads to be processsed in FIFO rather than LIFO  
+    /// by using TaskCreationOptions.preferFairness
+    /// </summary>
+    public TaskFactory TaskFactory { get; set; }
 }