Async logging queue

Its.Log.UnitTests/AsyncTests.cs

@@ -95,10 +95,7 @@ public class AsyncTestHelper
         public IAsyncResult BeginSomething(AsyncCallback callback, object state)
         {
             var activity = state as ILogActivity;
-            if (activity != null)
-            {
-                activity.Trace(() => "hello");
-            }
+            activity?.Trace(() => "hello");
 
             return new AsyncResult
                        {
@@ -111,14 +108,8 @@ public void EndSomething(IAsyncResult result)
             var state = result.AsyncState as Tuple<AsyncCallback, ILogActivity>;
             if (state != null)
             {
-                if (state.Item2 != null)
-                {
-                    state.Item2.Trace(() => "hello");
-                }
-                if (state.Item1 != null)
-                {
-                    state.Item1.Invoke(result);
-                }
+                state.Item2?.Trace(() => "hello");
+                state.Item1?.Invoke(result);
             }
         }
 

Its.Log.UnitTests/DispatchQueueTests.cs

@@ -0,0 +1,75 @@
+// Copyright (c) Microsoft. All rights reserved. 
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Collections.Generic;
+using FluentAssertions;
+using System.Linq;
+using System.Threading.Tasks;
+using Its.Log.Instrumentation.Extensions;
+using NUnit.Framework;
+
+namespace Its.Log.Instrumentation.UnitTests
+{
+    [TestFixture]
+    public class DispatchQueueTests
+    {
+        [Test]
+        public async Task log_entries_can_be_queued()
+        {
+            var log = new List<LogEntry>();
+            using (new LogEntryDispatchQueue(send: async e => log.Add(e)))
+            {
+                WriteInParallel(10);
+            }
+
+            log.Should().HaveCount(10);
+        }
+
+        [Test]
+        public async Task telemetry_events_can_be_queued()
+        {
+            var log = new List<Telemetry>();
+
+            using (new TelemetryDispatchQueue(send: async e => log.Add(e)))
+            {
+                WriteInParallel(10, _ =>
+                {
+                    using (Log.With<Telemetry>().Enter(() => { }))
+                    {
+                    }
+                });
+            }
+
+            log.Should().HaveCount(10);
+        }
+
+        [Test]
+        public async Task When_the_sender_throws_then_other_events_continue_to_be_sent()
+        {
+            var log = new List<LogEntry>();
+            var count = 0;
+
+            using (new LogEntryDispatchQueue(send: async e =>
+            {
+                if (++count == 1)
+                {
+                    throw new Exception("oops");
+                }
+                log.Add(e);
+            }))
+            {
+                WriteInParallel(10);
+            }
+
+            log.Should().HaveCount(9);
+        }
+
+        public void WriteInParallel(int degreesOfParallelism = 5, Action<int> write = null)
+        {
+            Parallel.ForEach(Enumerable.Range(1, degreesOfParallelism),
+                             write ??
+                             (i => Log.Write(() => i)));
+        }
+    }
+}

Its.Log.UnitTests/Its.Log.UnitTests.csproj

@@ -105,9 +105,6 @@
       <SpecificVersion>False</SpecificVersion>
       <HintPath>..\packages\Rx-PlatformServices.2.1.30214.0\lib\Net40\System.Reactive.PlatformServices.dll</HintPath>
     </Reference>
-    <Reference Include="System.ServiceModel">
-      <RequiredTargetFramework>3.0</RequiredTargetFramework>
-    </Reference>
     <Reference Include="System.Web" />
     <Reference Include="System.Web.Services" />
     <Reference Include="System.Windows.Forms" />
@@ -124,6 +121,7 @@
     <Compile Include="BoundaryTests.cs" />
     <Compile Include="AnonymousMethodInfoTests.cs" />
     <Compile Include="CounterTests.cs" />
+    <Compile Include="DispatchQueueTests.cs" />
     <Compile Include="ExceptionExtensionsTests.cs" />
     <Compile Include="ExceptionTrackingTests.cs" />
     <Compile Include="ExtensionTests.cs" />

Its.Log.UnitTests/TestHelper.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Linq;
 using System.Reactive.Linq;
+using Its.Log.Instrumentation.Extensions;
 
 namespace Its.Log.Instrumentation.UnitTests
 {
@@ -34,6 +35,12 @@ public static IDisposable SubscribeToLogInternalErrors(this IObserver<LogEntry>
             return subscription;
         }
 
+        public static IDisposable SubscribeToTelemetryEvents(this IObserver<Telemetry> observer)
+        {
+            var subscription = Log.TelemetryEvents().Subscribe(observer);
+            return subscription;
+        }
+
         public static IDisposable OnEntryPosted(Action<LogEntry> doSomething)
         {
             return Observable

Its.Log/AsyncDispatchQueue{T}.cs

@@ -0,0 +1,92 @@
+// Copyright (c) Microsoft. All rights reserved. 
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Collections.Concurrent;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Its.Log.Instrumentation
+{
+    public class AsyncDispatchQueue<T> : IDisposable
+    {
+        private readonly Func<T, Task> send;
+        private readonly IDisposable subscription;
+        private readonly BlockingCollection<T> blockingCollection;
+        private readonly Thread dispatcherThread;
+
+        protected AsyncDispatchQueue(IObservable<T> events, Func<T, Task> send)
+        {
+            if (send == null)
+            {
+                throw new ArgumentNullException(nameof(send));
+            }
+            if (events == null)
+            {
+                throw new ArgumentNullException(nameof(events));
+            }
+
+            this.send = send;
+            blockingCollection = new BlockingCollection<T>();
+            subscription = events.Subscribe(new Observer(blockingCollection));
+
+            dispatcherThread = new Thread(Send);
+            dispatcherThread.Start();
+        }
+
+        private void Send()
+        {
+            while (!blockingCollection.IsCompleted)
+            {
+                try
+                {
+                    var value = blockingCollection.Take();
+                    send(value).Wait();
+                }
+                catch (Exception exception)
+                {
+                    exception.RaiseErrorEvent();
+                }
+            }
+        }
+
+        private async Task Drain()
+        {
+            while (!blockingCollection.IsCompleted)
+            {
+                await Task.Delay(50);
+            }
+        }
+
+        public void Dispose()
+        {
+            blockingCollection.CompleteAdding();
+            subscription.Dispose();
+            Drain().Wait(TimeSpan.FromSeconds(30));
+        }
+
+        private class Observer : IObserver<T>
+        {
+            private readonly BlockingCollection<T> blockingCollection;
+
+            public Observer(BlockingCollection<T> blockingCollection)
+            {
+                if (blockingCollection == null)
+                {
+                    throw new ArgumentNullException(nameof(blockingCollection));
+                }
+                this.blockingCollection = blockingCollection;
+            }
+
+            public void OnNext(T value) => blockingCollection.Add(value);
+
+            public void OnError(Exception error)
+            {
+            }
+
+            public void OnCompleted()
+            {
+            }
+        }
+    }
+}

Its.Log/Its.Log.csproj

@@ -93,6 +93,7 @@
       <SubType>Code</SubType>
     </Compile>
     <Compile Include="AnonymousMethodInfo{T}.cs" />
+    <Compile Include="AsyncDispatchQueue{T}.cs" />
     <Compile Include="DefaultDiagnosticSensors.cs">
       <SubType>Code</SubType>
     </Compile>
@@ -126,10 +127,12 @@
     <Compile Include="Log.cs">
       <SubType>Code</SubType>
     </Compile>
+    <Compile Include="LogEntryDispatchQueue.cs" />
     <Compile Include="LogEntryExtensions.cs" />
     <Compile Include="LogEntry{T}.cs" />
     <Compile Include="MemberAccessor.cs" />
     <Compile Include="ObservableExtensions.cs" />
+    <Compile Include="TelemetryDispatchQueue.cs" />
     <Compile Include="TelemetryExtensions.cs" />
     <Compile Include="Text\ILogTextFormatter.cs" />
     <Compile Include="Text\MultiLineTextFormatter.cs" />

Its.Log/LogEntryDispatchQueue.cs

@@ -0,0 +1,18 @@
+// Copyright (c) Microsoft. All rights reserved. 
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Threading.Tasks;
+
+namespace Its.Log.Instrumentation
+{
+    public class LogEntryDispatchQueue : AsyncDispatchQueue<LogEntry>
+    {
+        public LogEntryDispatchQueue(
+            Func<LogEntry, Task> send,
+            IObservable<LogEntry> logEvents = null) :
+                base(logEvents ?? Log.Events(), send)
+        {
+        }
+    }
+}

Its.Log/TelemetryDispatchQueue.cs

@@ -0,0 +1,16 @@
+using System;
+using System.Threading.Tasks;
+using Its.Log.Instrumentation.Extensions;
+
+namespace Its.Log.Instrumentation
+{
+    public class TelemetryDispatchQueue : AsyncDispatchQueue<Telemetry>
+    {
+        public TelemetryDispatchQueue(
+            Func<Telemetry, Task> send,
+            IObservable<Telemetry> telemetryEvents = null) :
+                base(telemetryEvents ?? Log.TelemetryEvents(), send)
+        {
+        }
+    }
+}