Thoroughly test the sync event pipeline, couple tweaks

Author: MichaelGHSeg

Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs

@@ -1,5 +1,6 @@
 using System.Collections.Generic;
 using System.Threading;
+using System.Threading.Tasks;
 using global::System;
 using global::System.Linq;
 using Segment.Analytics.Policies;
@@ -70,9 +71,15 @@ public SyncEventPipeline(
         public void Put(RawEvent @event) => _writeChannel.Send(@event);
 
         public void Flush() {
-            FlushEvent flushEvent = new FlushEvent(new SemaphoreSlim(1,1));
+            FlushEvent flushEvent = new FlushEvent(new SemaphoreSlim(1));
+            // Set it up to be released by Upload
+            flushEvent._semaphore.Wait();
             _writeChannel.Send(flushEvent);
-            flushEvent._semaphore.Wait(_flushTimeout, _flushCancellationToken);
+            // Wait until the single slot in the semaphore is free. (if it's running!)
+            if (Running && !_uploadChannel.isCancelled)
+            {
+                flushEvent._semaphore.Wait(_flushTimeout, _flushCancellationToken);
+            }
         } 
 
         public void Start()
@@ -148,6 +155,7 @@ private void Upload() => _analytics.AnalyticsScope.Launch(_analytics.NetworkIODi
                 Analytics.Logger.Log(LogLevel.Debug, message: _logTag + " performing flush");
 
                 await Scope.WithContext(_analytics.FileIODispatcher, async () => await _storage.Rollover());
+                await Task.Delay(1); // The rollover might not quite be done, even 1 ms prevents an empty URL list
 
                 string[] fileUrlList = _storage.Read(StorageConstants.Events).Split(',');
                 foreach (string url in fileUrlList)

Tests/Utilities/EventPipelineTest.cs

@@ -10,6 +10,7 @@
 using Segment.Serialization;
 using Tests.Utils;
 using Xunit;
+using System.Linq;
 
 namespace Tests.Utilities
 {
@@ -225,6 +226,86 @@ public void TestConfigWithEventPipelineProviders(IEventPipelineProvider provider
             analytics.Track("test");
         }
 
+        [Fact]
+        public void TestSyncEventPipelineProviderWaits()
+        {
+            const int iterations = 100;
+            const int newAnalyticsEvery = 10;
+            const int eventCount = 10;
+
+            int totalTracks = 0;
+            int totalUploads = 0;
+
+            _mockHttpClient
+            .Setup(client => client.Upload(It.IsAny<byte[]>()))
+            .Callback<byte[]>(bytes =>
+            {
+                string content = System.Text.Encoding.UTF8.GetString(bytes);
+                int count = content.Split(new string[] { "test" }, StringSplitOptions.None).Length - 1;
+                totalUploads += count;
+            })
+            .ReturnsAsync(true);
+
+            var config = new Configuration(
+            writeKey: "123",
+            useSynchronizeDispatcher: true,
+            flushInterval: 100000,
+            flushAt: eventCount * 2,
+            httpClientProvider: new MockHttpClientProvider(_mockHttpClient),
+            storageProvider: new InMemoryStorageProvider(),
+            eventPipelineProvider: new SyncEventPipelineProvider()
+            );
+
+            var analytics = new Analytics(config);
+            for (int j = 0; j < iterations; j++) 
+            {
+                if (j % newAnalyticsEvery == 0)
+                {
+                    analytics = new Analytics(config);
+                }
+                _mockHttpClient.Invocations.Clear();
+                for (int i = 0; i < eventCount; i++)
+                {
+                    analytics.Track($"test {i}");
+                    totalTracks++;
+                }
+                analytics.Flush();
+
+#pragma warning disable CS4014 // Silly compiler, this isn't an invocation so it doesn't need to be awaited
+                _mockHttpClient.Verify(client => client.Upload(It.IsAny<byte[]>()), Times.AtLeastOnce, $"Iteration {j} of {eventCount}");
+#pragma warning restore CS4014 
+                IInvocation lastUploadInvocation = _mockHttpClient.Invocations.Last(invocation => invocation.Method.Name == "Upload");
+                int testsUploaded = System.Text.Encoding.UTF8
+                    .GetString((byte[])lastUploadInvocation.Arguments[0])
+                    .Split(new string[] { "test" }, StringSplitOptions.None).Length - 1;
+                Assert.Equal(eventCount, testsUploaded);
+            }
+            Assert.Equal(totalTracks, totalUploads);
+        }
+
+        [Fact]
+        public void TestRepeatedFlushesDontHang()
+        {
+            var config = new Configuration(
+                writeKey: "123",
+                useSynchronizeDispatcher: true,
+                flushInterval: 0,
+                flushAt: 1,
+                httpClientProvider: new MockHttpClientProvider(_mockHttpClient),
+                storageProvider: new MockStorageProvider(_storage),
+                eventPipelineProvider: new SyncEventPipelineProvider(5000)
+            );
+            var analytics = new Analytics(config);
+            analytics.Track("test");
+            DateTime startTime = DateTime.Now;
+            analytics.Flush();
+            analytics.Flush();
+            analytics.Flush();
+            analytics.Flush();
+            analytics.Flush();
+            Assert.True(DateTime.Now - startTime < TimeSpan.FromMilliseconds(100));
+        }
+
         [Fact]
         public void TestConfigWithCustomEventPipelineProvider()
         {
@@ -248,13 +329,9 @@ public void TestConfigWithCustomEventPipelineProvider()
 
         public class CustomEventPipelineProvider : IEventPipelineProvider
         {
-            public CustomEventPipelineProvider()
-            {
-            }
-
+            public CustomEventPipelineProvider() {}
             public IEventPipeline Create(Analytics analytics, string key)
             {
-                // Custom implementation
                 return new CustomEventPipeline(analytics, key);
             }