[Do not merged]Buffered Writes to rpc channel and async everywhere

src/Eventing/EventSources.cs

@@ -0,0 +1,13 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+namespace Microsoft.Azure.Functions.PowerShellWorker
+{
+    internal static class EventSources
+    {
+        public const string ScriptFiles = "ScriptFiles";
+        public const string Rpc = "Rpc";
+        public const string Worker = "Worker";
+        public const string WorkerProcess = "WorkerProcess";
+    }
+}

src/Eventing/IScriptEventManager.cs

@@ -0,0 +1,12 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+
+namespace Microsoft.Azure.Functions.PowerShellWorker
+{
+    internal interface IScriptEventManager : IObservable<ScriptEvent>
+    {
+        void Publish(ScriptEvent scriptEvent);
+    }
+}

src/Eventing/Rpc/InboundEvent.cs

@@ -0,0 +1,17 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using Grpc.Core;
+using Microsoft.Azure.WebJobs.Script.Grpc.Messages;
+
+namespace Microsoft.Azure.Functions.PowerShellWorker
+{
+    internal class InboundEvent : RpcEvent
+    {
+        public IAsyncStreamReader<StreamingMessage> requestStream;
+
+        public InboundEvent(string workerId, StreamingMessage message) : base(workerId, message, MessageOrigin.Worker)
+        {
+        }
+    }
+}

src/Eventing/Rpc/OutboundEvent.cs

@@ -0,0 +1,13 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+using Microsoft.Azure.WebJobs.Script.Grpc.Messages;
+
+namespace Microsoft.Azure.Functions.PowerShellWorker
+{
+    internal class OutboundEvent : RpcEvent
+    {
+        public OutboundEvent(string workerId, StreamingMessage message) : base(workerId, message, MessageOrigin.Host)
+        {
+        }
+    }
+}

src/Eventing/Rpc/RpcEvent.cs

@@ -0,0 +1,31 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+using Microsoft.Azure.WebJobs.Script.Grpc.Messages;
+
+namespace Microsoft.Azure.Functions.PowerShellWorker
+{
+    internal class RpcEvent : ScriptEvent
+    {
+        internal RpcEvent(string workerId, StreamingMessage message, MessageOrigin origin = MessageOrigin.Host)
+            : base(message.ContentCase.ToString(), EventSources.Rpc)
+        {
+            Message = message;
+            Origin = origin;
+            WorkerId = workerId;
+        }
+
+        public enum MessageOrigin
+        {
+            Worker,
+            Host
+        }
+
+        public MessageOrigin Origin { get; }
+
+        public StreamingMessage.ContentOneofCase MessageType => Message.ContentCase;
+
+        public string WorkerId { get; }
+
+        public StreamingMessage Message { get; }
+    }
+}

src/Eventing/ScriptEvent.cs

@@ -0,0 +1,18 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+namespace Microsoft.Azure.Functions.PowerShellWorker
+{
+    internal class ScriptEvent
+    {
+        public ScriptEvent(string name, string source)
+        {
+            Name = name;
+            Source = source;
+        }
+
+        public string Name { get; }
+
+        public string Source { get; }
+    }
+}

src/Eventing/ScriptEventManager.cs

@@ -0,0 +1,51 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using System.Reactive.Subjects;
+
+namespace Microsoft.Azure.Functions.PowerShellWorker
+{
+    internal sealed class ScriptEventManager : IScriptEventManager, IDisposable
+    {
+        private readonly Subject<ScriptEvent> _subject = new Subject<ScriptEvent>();
+        private bool _disposed = false;
+
+        public void Publish(ScriptEvent scriptEvent)
+        {
+            ThrowIfDisposed();
+
+            _subject.OnNext(scriptEvent);
+        }
+
+        public IDisposable Subscribe(IObserver<ScriptEvent> observer)
+        {
+            ThrowIfDisposed();
+
+            return _subject.Subscribe(observer);
+        }
+
+        private void ThrowIfDisposed()
+        {
+            if (_disposed)
+            {
+                throw new ObjectDisposedException(nameof(ScriptEventManager));
+            }
+        }
+
+        private void Dispose(bool disposing)
+        {
+            if (!_disposed)
+            {
+                if (disposing)
+                {
+                    _subject.Dispose();
+                }
+
+                _disposed = true;
+            }
+        }
+
+        public void Dispose() => Dispose(true);
+    }
+}

src/Logging/RpcLogger.cs

@@ -55,7 +55,7 @@ public void Log(LogLevel logLevel, string message, Exception exception = null, b
                     }
                 };
 
-                _msgStream.Write(logMessage);
+                _msgStream.AddToBlockingQueue(logMessage);
             }
             else
             {

src/Messaging/MessagingStream.cs

@@ -4,6 +4,7 @@
 //
 
 using System;
+using System.Collections.Concurrent;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -15,7 +16,7 @@ namespace Microsoft.Azure.Functions.PowerShellWorker.Messaging
     internal class MessagingStream
     {
         private readonly AsyncDuplexStreamingCall<StreamingMessage, StreamingMessage> _call;
-        private readonly SemaphoreSlim _semaphoreSlim = new SemaphoreSlim(initialCount: 1, maxCount: 1);
+        private BlockingCollection<StreamingMessage> _blockingCollectionQueue = new BlockingCollection<StreamingMessage>();
 
         internal MessagingStream(string host, int port)
         {
@@ -28,30 +29,34 @@ internal MessagingStream(string host, int port)
         /// </summary>
         internal StreamingMessage GetCurrentMessage() => _call.ResponseStream.Current;
 
+        internal void AddToBlockingQueue(StreamingMessage streamingMessage)
+        {
+            _blockingCollectionQueue.Add(streamingMessage);
+        }
+
         /// <summary>
         /// Move to the next message.
         /// </summary>
         internal async Task<bool> MoveNext() => await _call.ResponseStream.MoveNext(CancellationToken.None);
 
-        /// <summary>
-        /// Write the outgoing message.
-        /// </summary>
-        internal void Write(StreamingMessage message) => WriteImplAsync(message).ConfigureAwait(false);
-
         /// <summary>
         /// Take a message from the buffer and write to the gRPC channel.
         /// </summary>
-        private async Task WriteImplAsync(StreamingMessage message)
+        internal Task Write()
         {
-            try
-            {
-                await _semaphoreSlim.WaitAsync();
-                await _call.RequestStream.WriteAsync(message);
-            }
-            finally
+            var consumer = Task.Run(async () =>
             {
-                _semaphoreSlim.Release();
-            }
+                foreach (var rpcWriteMsg in _blockingCollectionQueue.GetConsumingEnumerable())
+                {
+                    await _call.RequestStream.WriteAsync(rpcWriteMsg);
+                }
+            });
+            return consumer;
+        }
+
+        internal Task Write(StreamingMessage msg)
+        {
+           return _call.RequestStream.WriteAsync(msg);
         }
     }
 }

src/Microsoft.Azure.Functions.PowerShellWorker.csproj

@@ -25,6 +25,7 @@ Licensed under the MIT license. See LICENSE file in the project root for full li
     <PackageReference Include="Newtonsoft.Json" Version="12.0.1" />
     <PackageReference Include="CommandLineParser" Version="2.3.0" />
     <PackageReference Include="Google.Protobuf" Version="3.6.1" />
+    <PackageReference Include="System.Reactive" Version="4.1.5" />
   </ItemGroup>
 
   <ItemGroup>