From 911370ddf0a35255db76a1d085ec8f4b79148c64 Mon Sep 17 00:00:00 2001
From: Michael Grosse Huelsewiesche <mihuelsewiesche@twilio.com>
Date: Mon, 29 Apr 2024 12:24:24 -0400
Subject: [PATCH 1/4] Provider pattern for EventPipleline and new
 SyncEventPipeline for synchronous flush

---
 .../Segment/Analytics/Configuration.cs        |   6 +-
 .../Analytics/Plugins/SegmentDestination.cs   |  10 +-
 .../Analytics/Utilities/EventPipeline.cs      |   2 +-
 .../Utilities/EventPipelineProvider.cs        |  17 ++
 .../Analytics/Utilities/IEventPipeline.cs     |  13 ++
 .../Utilities/IEventPipelineProvider.cs       |   9 +
 .../Analytics/Utilities/SyncEventPipeline.cs  | 204 ++++++++++++++++++
 .../Utilities/SyncEventPipelineProvider.cs    |  28 +++
 8 files changed, 279 insertions(+), 10 deletions(-)
 create mode 100644 Analytics-CSharp/Segment/Analytics/Utilities/EventPipelineProvider.cs
 create mode 100644 Analytics-CSharp/Segment/Analytics/Utilities/IEventPipeline.cs
 create mode 100644 Analytics-CSharp/Segment/Analytics/Utilities/IEventPipelineProvider.cs
 create mode 100644 Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs
 create mode 100644 Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs

diff --git a/Analytics-CSharp/Segment/Analytics/Configuration.cs b/Analytics-CSharp/Segment/Analytics/Configuration.cs
index c492efe..8f51743 100644
--- a/Analytics-CSharp/Segment/Analytics/Configuration.cs
+++ b/Analytics-CSharp/Segment/Analytics/Configuration.cs
@@ -45,6 +45,8 @@ private set
 
         public IList<IFlushPolicy> FlushPolicies { get; }
 
+        public EventPipelineProvider EventPipelineProvider { get; }
+
         /// <summary>
         /// Configuration that analytics can use
         /// </summary>
@@ -82,7 +84,8 @@ public Configuration(string writeKey,
             IAnalyticsErrorHandler analyticsErrorHandler = null,
             IStorageProvider storageProvider = default,
             IHTTPClientProvider httpClientProvider = default,
-            IList<IFlushPolicy> flushPolicies = default)
+            IList<IFlushPolicy> flushPolicies = default,
+            EventPipelineProvider eventPipelineProvider = default)
         {
             WriteKey = writeKey;
             FlushAt = flushAt;
@@ -98,6 +101,7 @@ public Configuration(string writeKey,
             FlushPolicies = flushPolicies == null ? new ConcurrentList<IFlushPolicy>() : new ConcurrentList<IFlushPolicy>(flushPolicies);
             FlushPolicies.Add(new CountFlushPolicy(flushAt));
             FlushPolicies.Add(new FrequencyFlushPolicy(flushInterval * 1000L));
+            EventPipelineProvider = eventPipelineProvider ?? new EventPipelineProvider();
         }
 
         public Configuration(string writeKey,
diff --git a/Analytics-CSharp/Segment/Analytics/Plugins/SegmentDestination.cs b/Analytics-CSharp/Segment/Analytics/Plugins/SegmentDestination.cs
index 0c5db70..9586be0 100644
--- a/Analytics-CSharp/Segment/Analytics/Plugins/SegmentDestination.cs
+++ b/Analytics-CSharp/Segment/Analytics/Plugins/SegmentDestination.cs
@@ -15,7 +15,7 @@ namespace Segment.Analytics.Plugins
     /// </summary>
     public class SegmentDestination : DestinationPlugin, ISubscriber
     {
-        private EventPipeline _pipeline = null;
+        private IEventPipeline _pipeline = null;
 
         public override string Key => "Segment.io";
 
@@ -64,13 +64,7 @@ public override void Configure(Analytics analytics)
             // Add DestinationMetadata enrichment plugin
             Add(new DestinationMetadataPlugin());
 
-            _pipeline = new EventPipeline(
-                    analytics,
-                    Key,
-                    analytics.Configuration.WriteKey,
-                    analytics.Configuration.FlushPolicies,
-                    analytics.Configuration.ApiHost
-                );
+            _pipeline = analytics.Configuration.EventPipelineProvider.Create(analytics, Key);
 
             analytics.AnalyticsScope.Launch(analytics.AnalyticsDispatcher, async () =>
             {
diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/EventPipeline.cs b/Analytics-CSharp/Segment/Analytics/Utilities/EventPipeline.cs
index e9dd4fc..c54a18f 100644
--- a/Analytics-CSharp/Segment/Analytics/Utilities/EventPipeline.cs
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/EventPipeline.cs
@@ -7,7 +7,7 @@
 
 namespace Segment.Analytics.Utilities
 {
-    internal class EventPipeline
+    public class EventPipeline: IEventPipeline
     {
         private readonly Analytics _analytics;
 
diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/EventPipelineProvider.cs b/Analytics-CSharp/Segment/Analytics/Utilities/EventPipelineProvider.cs
new file mode 100644
index 0000000..abd376c
--- /dev/null
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/EventPipelineProvider.cs
@@ -0,0 +1,17 @@
+namespace Segment.Analytics.Utilities
+{
+    public class EventPipelineProvider:IEventPipelineProvider
+    {
+        public EventPipelineProvider()
+        {
+        }
+
+        public IEventPipeline Create(Analytics analytics, string key)
+        {
+            return new EventPipeline(analytics, key, 
+                    analytics.Configuration.WriteKey,
+                    analytics.Configuration.FlushPolicies,
+                    analytics.Configuration.ApiHost);
+        }
+    }
+}
\ No newline at end of file
diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/IEventPipeline.cs b/Analytics-CSharp/Segment/Analytics/Utilities/IEventPipeline.cs
new file mode 100644
index 0000000..b9cce6d
--- /dev/null
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/IEventPipeline.cs
@@ -0,0 +1,13 @@
+namespace Segment.Analytics.Utilities
+{
+    public interface IEventPipeline
+    {
+        bool Running { get; }
+        string ApiHost { get; set; }
+
+        void Put(RawEvent @event);
+        void Flush();
+        void Start();
+        void Stop();
+    }
+}
\ No newline at end of file
diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/IEventPipelineProvider.cs b/Analytics-CSharp/Segment/Analytics/Utilities/IEventPipelineProvider.cs
new file mode 100644
index 0000000..2e39b2b
--- /dev/null
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/IEventPipelineProvider.cs
@@ -0,0 +1,9 @@
+using System.Collections.Generic;
+
+namespace Segment.Analytics.Utilities
+{
+    public interface IEventPipelineProvider
+    {
+        IEventPipeline Create(Analytics analytics, string key);
+    }
+}
\ No newline at end of file
diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs
new file mode 100644
index 0000000..6ff5bdd
--- /dev/null
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs
@@ -0,0 +1,204 @@
+using System.Collections.Generic;
+using System.Threading;
+using global::System;
+using global::System.Linq;
+using Segment.Analytics.Policies;
+using Segment.Concurrent;
+using Segment.Serialization;
+
+namespace Segment.Analytics.Utilities
+{
+    public class SyncEventPipeline: IEventPipeline
+    {
+        private readonly Analytics _analytics;
+
+        private readonly string _logTag;
+
+        private readonly IList<IFlushPolicy> _flushPolicies;
+
+        private Channel<RawEvent> _writeChannel;
+
+        private Channel<string> _uploadChannel;
+        private Channel<string> _flushCompleteChannel;
+
+        private readonly HTTPClient _httpClient;
+
+        private readonly IStorage _storage;
+
+        internal string ApiHost { get; set; }
+
+        public bool Running { get; private set; }
+
+        internal const string FlushPoison = "#!flush";
+
+        internal static readonly RawEvent s_flushEvent = new ScreenEvent(FlushPoison, FlushPoison);
+
+        internal const string UploadSig = "#!upload";
+        internal const string FlushCompleteSig = "#!complete";
+        internal int _flushTimeout = 5000;
+        internal CancellationToken? _flushCancellationToken = null;
+
+        public SyncEventPipeline(
+            Analytics analytics,
+            string logTag,
+            string apiKey,
+            IList<IFlushPolicy> flushPolicies,
+            string apiHost = HTTPClient.DefaultAPIHost,
+            int flushTimeout = 5000,
+            CancellationToken? flushCancellationToken = null)
+        {
+            _analytics = analytics;
+            _logTag = logTag;
+            _flushPolicies = flushPolicies;
+            ApiHost = apiHost;
+
+            _writeChannel = new Channel<RawEvent>();
+            _uploadChannel = new Channel<string>();
+            _flushCompleteChannel = null;
+            _httpClient = analytics.Configuration.HttpClientProvider.CreateHTTPClient(apiKey, apiHost: apiHost);
+            _httpClient.AnalyticsRef = analytics;
+            _storage = analytics.Storage;
+            Running = false;
+            _flushTimeout = flushTimeout;
+            _flushCancellationToken = flushCancellationToken;
+        }
+
+        public void Put(RawEvent @event) => _writeChannel.Send(@event);
+
+        public void Flush() {
+            _flushCompleteChannel = new Channel<string>();
+            _writeChannel.Send(s_flushEvent);
+            _flushCompleteChannel.Receive().Wait(_flushTimeout,_flushCancellationToken??CancellationToken.None);
+            _flushCompleteChannel = null;
+        } 
+
+        public void Start()
+        {
+            if (Running) return;
+
+            // avoid to re-establish a channel if the pipeline just gets created
+            if (_writeChannel.isCancelled)
+            {
+                _writeChannel = new Channel<RawEvent>();
+                _uploadChannel = new Channel<string>();
+                _flushCompleteChannel = new Channel<string>();
+            }
+
+            Running = true;
+            Schedule();
+            Write();
+            Upload();
+        }
+
+        public void Stop()
+        {
+            if (!Running) return;
+            Running = false;
+
+            _uploadChannel.Cancel();
+            _flushCompleteChannel.Cancel();
+            _writeChannel.Cancel();
+            Unschedule();
+        }
+
+        private void Write() => _analytics.AnalyticsScope.Launch(_analytics.FileIODispatcher, async () =>
+        {
+            while (!_writeChannel.isCancelled)
+            {
+                RawEvent e = await _writeChannel.Receive();
+                bool isPoison = e == s_flushEvent;
+
+                if (!isPoison)
+                {
+                    try
+                    {
+                        string str = JsonUtility.ToJson(e);
+                        Analytics.Logger.Log(LogLevel.Debug, message: _logTag + " running " + str);
+                        await _storage.Write(StorageConstants.Events, str);
+
+                        foreach (IFlushPolicy flushPolicy in _flushPolicies)
+                        {
+                            flushPolicy.UpdateState(e);
+                        }
+                    }
+                    catch (Exception exception)
+                    {
+                        Analytics.Logger.Log(LogLevel.Error, exception, _logTag + ": Error writing events to storage.");
+                    }
+                }
+
+                if (isPoison || _flushPolicies.Any(o => o.ShouldFlush()))
+                {
+                    _uploadChannel.Send(UploadSig);
+                    foreach (IFlushPolicy flushPolicy in _flushPolicies)
+                    {
+                        flushPolicy.Reset();
+                    }
+                }
+            }
+        });
+
+        private void Upload() => _analytics.AnalyticsScope.Launch(_analytics.NetworkIODispatcher, async () =>
+        {
+            while (!_uploadChannel.isCancelled)
+            {
+                bool reportFlushComplete = _flushCompleteChannel != null;
+                await _uploadChannel.Receive();
+                Analytics.Logger.Log(LogLevel.Debug, message: _logTag + " performing flush");
+
+                await Scope.WithContext(_analytics.FileIODispatcher, async () => await _storage.Rollover());
+
+                string[] fileUrlList = _storage.Read(StorageConstants.Events).Split(',');
+                foreach (string url in fileUrlList)
+                {
+                    if (string.IsNullOrEmpty(url))
+                    {
+                        continue;
+                    }
+
+                    byte[] data = _storage.ReadAsBytes(url);
+                    if (data == null)
+                    {
+                        continue;
+                    }
+
+                    bool shouldCleanup = true;
+                    try
+                    {
+                        shouldCleanup = await _httpClient.Upload(data);
+                        Analytics.Logger.Log(LogLevel.Debug, message: _logTag + " uploaded " + url);
+                    }
+                    catch (Exception e)
+                    {
+                        Analytics.Logger.Log(LogLevel.Error, e, _logTag + ": Error uploading to url");
+                    }
+
+                    if (shouldCleanup)
+                    {
+                        _storage.RemoveFile(url);
+                    }
+                }
+                if(reportFlushComplete && _flushCompleteChannel != null)
+                {
+                    _flushCompleteChannel.Send(FlushCompleteSig);
+                }
+            }
+        });
+
+        private void Schedule()
+        {
+            foreach (IFlushPolicy flushPolicy in _flushPolicies)
+            {
+                flushPolicy.Schedule(_analytics);
+            }
+        }
+
+        private void Unschedule()
+        {
+            foreach (IFlushPolicy flushPolicy in _flushPolicies)
+            {
+                flushPolicy.Unschedule();
+            }
+        }
+    }
+}
diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs
new file mode 100644
index 0000000..65cf1dc
--- /dev/null
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs
@@ -0,0 +1,28 @@
+using System.Threading;
+
+namespace Segment.Analytics.Utilities
+{
+    public class SyncEventPipelineProvider: IEventPipelineProvider
+    {
+        internal int _flushTimeout = 5000;
+        internal CancellationToken? _flushCancellationToken = null;
+
+        public SyncEventPipelineProvider(
+            int flushTimeout = 5000,
+            CancellationToken? flushCancellationToken = null)
+        {
+            _flushTimeout = flushTimeout;
+            _flushCancellationToken = flushCancellationToken;
+        }
+
+        public IEventPipeline Create(Analytics analytics, string key)
+        {
+            return new SyncEventPipeline(analytics, key, 
+                    analytics.Configuration.WriteKey,
+                    analytics.Configuration.FlushPolicies,
+                    analytics.Configuration.ApiHost,
+                    _flushTimeout,
+                    _flushCancellationToken);
+        }
+    }
+}
\ No newline at end of file

From 841a9b65dd2fb4098e3bf6895f941d76e3398084 Mon Sep 17 00:00:00 2001
From: Michael Grosse Huelsewiesche <mihuelsewiesche@twilio.com>
Date: Mon, 29 Apr 2024 19:22:01 -0400
Subject: [PATCH 2/4] Converting to use a semaphore flush event to better
 handle concurrent flush calls.

---
 .../Segment/Analytics/Configuration.cs        |  2 +-
 .../Analytics/Utilities/EventPipeline.cs      |  2 +-
 .../Analytics/Utilities/SyncEventPipeline.cs  | 57 +++++++++----------
 .../Utilities/SyncEventPipelineProvider.cs    |  3 +-
 4 files changed, 29 insertions(+), 35 deletions(-)

diff --git a/Analytics-CSharp/Segment/Analytics/Configuration.cs b/Analytics-CSharp/Segment/Analytics/Configuration.cs
index 8f51743..d8d997b 100644
--- a/Analytics-CSharp/Segment/Analytics/Configuration.cs
+++ b/Analytics-CSharp/Segment/Analytics/Configuration.cs
@@ -45,7 +45,7 @@ private set
 
         public IList<IFlushPolicy> FlushPolicies { get; }
 
-        public EventPipelineProvider EventPipelineProvider { get; }
+        public IEventPipelineProvider EventPipelineProvider { get; }
 
         /// <summary>
         /// Configuration that analytics can use
diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/EventPipeline.cs b/Analytics-CSharp/Segment/Analytics/Utilities/EventPipeline.cs
index c54a18f..21fe7ce 100644
--- a/Analytics-CSharp/Segment/Analytics/Utilities/EventPipeline.cs
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/EventPipeline.cs
@@ -23,7 +23,7 @@ public class EventPipeline: IEventPipeline
 
         private readonly IStorage _storage;
 
-        internal string ApiHost { get; set; }
+        public string ApiHost { get; set; }
 
         public bool Running { get; private set; }
 
diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs
index 6ff5bdd..f8b18ee 100644
--- a/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs
@@ -8,6 +8,18 @@
 
 namespace Segment.Analytics.Utilities
 {
+    internal sealed class FlushEvent : RawEvent
+    {
+        public override string Type => "flush";
+        public readonly Semaphore _semaphore;
+
+        internal FlushEvent(Semaphore semaphore)
+        {
+            _semaphore = semaphore;
+        }
+   }
+
+
     public class SyncEventPipeline: IEventPipeline
     {
         private readonly Analytics _analytics;
@@ -18,25 +30,17 @@ public class SyncEventPipeline: IEventPipeline
 
         private Channel<RawEvent> _writeChannel;
 
-        private Channel<string> _uploadChannel;
-        private Channel<string> _flushCompleteChannel;
+        private Channel<FlushEvent> _uploadChannel;
 
         private readonly HTTPClient _httpClient;
 
         private readonly IStorage _storage;
 
-        internal string ApiHost { get; set; }
+        public string ApiHost { get; set; }
 
         public bool Running { get; private set; }
 
-        internal const string FlushPoison = "#!flush";
-
-        internal static readonly RawEvent s_flushEvent = new ScreenEvent(FlushPoison, FlushPoison);
-
-        internal const string UploadSig = "#!upload";
-        internal const string FlushCompleteSig = "#!complete";
-        internal int _flushTimeout = 5000;
-        internal CancellationToken? _flushCancellationToken = null;
+        internal int _flushTimeout = -1;
 
         public SyncEventPipeline(
             Analytics analytics,
@@ -44,8 +48,7 @@ public SyncEventPipeline(
             string apiKey,
             IList<IFlushPolicy> flushPolicies,
             string apiHost = HTTPClient.DefaultAPIHost,
-            int flushTimeout = 5000,
-            CancellationToken? flushCancellationToken = null)
+            int flushTimeout = -1)
         {
             _analytics = analytics;
             _logTag = logTag;
@@ -53,23 +56,20 @@ public SyncEventPipeline(
             ApiHost = apiHost;
 
             _writeChannel = new Channel<RawEvent>();
-            _uploadChannel = new Channel<string>();
-            _flushCompleteChannel = null;
+            _uploadChannel = new Channel<FlushEvent>();
             _httpClient = analytics.Configuration.HttpClientProvider.CreateHTTPClient(apiKey, apiHost: apiHost);
             _httpClient.AnalyticsRef = analytics;
             _storage = analytics.Storage;
             Running = false;
             _flushTimeout = flushTimeout;
-            _flushCancellationToken = flushCancellationToken;
         }
 
         public void Put(RawEvent @event) => _writeChannel.Send(@event);
 
         public void Flush() {
-            _flushCompleteChannel = new Channel<string>();
-            _writeChannel.Send(s_flushEvent);
-            _flushCompleteChannel.Receive().Wait(_flushTimeout,_flushCancellationToken??CancellationToken.None);
-            _flushCompleteChannel = null;
+            FlushEvent flushEvent = new FlushEvent(new Semaphore(1,1));
+            _writeChannel.Send(flushEvent);
+            flushEvent._semaphore.WaitOne(_flushTimeout);
         } 
 
         public void Start()
@@ -80,8 +80,7 @@ public void Start()
             if (_writeChannel.isCancelled)
             {
                 _writeChannel = new Channel<RawEvent>();
-                _uploadChannel = new Channel<string>();
-                _flushCompleteChannel = new Channel<string>();
+                _uploadChannel = new Channel<FlushEvent>();
             }
 
             Running = true;
@@ -96,7 +95,6 @@ public void Stop()
             Running = false;
 
             _uploadChannel.Cancel();
-            _flushCompleteChannel.Cancel();
             _writeChannel.Cancel();
             Unschedule();
         }
@@ -106,7 +104,7 @@ private void Write() => _analytics.AnalyticsScope.Launch(_analytics.FileIODispat
             while (!_writeChannel.isCancelled)
             {
                 RawEvent e = await _writeChannel.Receive();
-                bool isPoison = e == s_flushEvent;
+                bool isPoison = e is FlushEvent;
 
                 if (!isPoison)
                 {
@@ -129,7 +127,8 @@ private void Write() => _analytics.AnalyticsScope.Launch(_analytics.FileIODispat
 
                 if (isPoison || _flushPolicies.Any(o => o.ShouldFlush()))
                 {
-                    _uploadChannel.Send(UploadSig);
+                    FlushEvent flushEvent = e as FlushEvent ?? new FlushEvent(null);
+                    _uploadChannel.Send(flushEvent);
                     foreach (IFlushPolicy flushPolicy in _flushPolicies)
                     {
                         flushPolicy.Reset();
@@ -142,8 +141,7 @@ private void Upload() => _analytics.AnalyticsScope.Launch(_analytics.NetworkIODi
         {
             while (!_uploadChannel.isCancelled)
             {
-                bool reportFlushComplete = _flushCompleteChannel != null;
-                await _uploadChannel.Receive();
+                FlushEvent flushEvent = await _uploadChannel.Receive();
                 Analytics.Logger.Log(LogLevel.Debug, message: _logTag + " performing flush");
 
                 await Scope.WithContext(_analytics.FileIODispatcher, async () => await _storage.Rollover());
@@ -178,10 +176,7 @@ private void Upload() => _analytics.AnalyticsScope.Launch(_analytics.NetworkIODi
                         _storage.RemoveFile(url);
                     }
                 }
-                if(reportFlushComplete && _flushCompleteChannel != null)
-                {
-                    _flushCompleteChannel.Send(FlushCompleteSig);
-                }
+                flushEvent._semaphore?.Release();
             }
         });
 
diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs
index 65cf1dc..17f5373 100644
--- a/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs
@@ -21,8 +21,7 @@ public IEventPipeline Create(Analytics analytics, string key)
                     analytics.Configuration.WriteKey,
                     analytics.Configuration.FlushPolicies,
                     analytics.Configuration.ApiHost,
-                    _flushTimeout,
-                    _flushCancellationToken);
+                    _flushTimeout);
         }
     }
 }
\ No newline at end of file

From d1000344e1653aaebae7c6b9c27f5ab3fbfa2a4f Mon Sep 17 00:00:00 2001
From: Michael Grosse Huelsewiesche <mihuelsewiesche@twilio.com>
Date: Mon, 29 Apr 2024 20:00:35 -0400
Subject: [PATCH 3/4] Modifying EventPipelineTest to test multiple types

---
 Tests/Utilities/EventPipelineTest.cs | 106 +++++++++++++++------------
 1 file changed, 59 insertions(+), 47 deletions(-)

diff --git a/Tests/Utilities/EventPipelineTest.cs b/Tests/Utilities/EventPipelineTest.cs
index 2ac8d3d..02d20d2 100644
--- a/Tests/Utilities/EventPipelineTest.cs
+++ b/Tests/Utilities/EventPipelineTest.cs
@@ -1,4 +1,6 @@
 using System;
+using System.Collections;
+using System.Collections.Generic;
 using System.Threading.Tasks;
 using Moq;
 using Segment.Analytics;
@@ -13,8 +15,6 @@ namespace Tests.Utilities
 {
     public class EventPipelineTest
     {
-        private EventPipeline _eventPipeline;
-
         private readonly Analytics _analytics;
 
         private readonly Mock<IStorage> _storage;
@@ -25,6 +25,12 @@ public class EventPipelineTest
 
         private readonly byte[] _bytes;
 
+        public static IEnumerable<object[]> GetPipelineProvider()
+        {
+            yield return new object[] { new EventPipelineProvider() };
+            yield return new object[] { new SyncEventPipelineProvider() };
+        }
+
         public EventPipelineTest()
         {
             Settings? settings = JsonUtility.FromJson<Settings?>(
@@ -50,14 +56,6 @@ public EventPipelineTest()
                 storageProvider: new MockStorageProvider(_storage)
             );
             _analytics = new Analytics(config);
-            _eventPipeline = new EventPipeline(
-                _analytics,
-                logTag: "key",
-                apiKey: _analytics.Configuration.WriteKey,
-                flushPolicies: _analytics.Configuration.FlushPolicies,
-                apiHost: _analytics.Configuration.ApiHost
-            );
-
             _file = Guid.NewGuid().ToString();
             _bytes = _file.GetBytes();
             _storage
@@ -68,23 +66,27 @@ public EventPipelineTest()
                 .Returns(_bytes);
         }
 
-        [Fact]
-        public async Task TestPut()
+        [Theory]
+        [MemberData(nameof(GetPipelineProvider))]
+        public async Task TestPut(IEventPipelineProvider provider)
         {
-            _eventPipeline.Start();
-            _eventPipeline.Put(new ScreenEvent("test"));
+            IEventPipeline eventPipeline = provider.Create(_analytics, "key");
+            eventPipeline.Start();
+            eventPipeline.Put(new ScreenEvent("test"));
 
             await Task.Delay(1000);
 
             _storage.Verify(o => o.Write(StorageConstants.Events, It.IsAny<string>()), Times.Exactly(1));
         }
 
-        [Fact]
-        public async Task TestFlush()
+        [Theory]
+        [MemberData(nameof(GetPipelineProvider))]
+        public async Task TestFlush(IEventPipelineProvider provider)
         {
-            _eventPipeline.Start();
-            _eventPipeline.Put(new ScreenEvent("test"));
-            _eventPipeline.Flush();
+            IEventPipeline eventPipeline = provider.Create(_analytics, "key");
+            eventPipeline.Start();
+            eventPipeline.Put(new ScreenEvent("test"));
+            eventPipeline.Flush();
 
             await Task.Delay(1000);
 
@@ -94,28 +96,32 @@ public async Task TestFlush()
             _storage.Verify(o => o.RemoveFile(_file), Times.Exactly(1));
         }
 
-        [Fact]
-        public void TestStart()
+        [Theory]
+        [MemberData(nameof(GetPipelineProvider))]
+        public void TestStart(IEventPipelineProvider provider)
         {
-            _eventPipeline.Start();
-            Assert.True(_eventPipeline.Running);
+            IEventPipeline eventPipeline = provider.Create(_analytics, "key");
+            eventPipeline.Start();
+            Assert.True(eventPipeline.Running);
         }
 
-        [Fact]
-        public async void TestStop()
+        [Theory]
+        [MemberData(nameof(GetPipelineProvider))]
+        public async void TestStop(IEventPipelineProvider provider)
         {
-            _eventPipeline.Start();
-            Assert.True(_eventPipeline.Running);
-            _eventPipeline.Stop();
-            Assert.False(_eventPipeline.Running);
+            IEventPipeline eventPipeline = provider.Create(_analytics, "key");
+            eventPipeline.Start();
+            Assert.True(eventPipeline.Running);
+            eventPipeline.Stop();
+            Assert.False(eventPipeline.Running);
 
             // make sure writeChannel is stopped
-            _eventPipeline.Put(new ScreenEvent("test"));
+            eventPipeline.Put(new ScreenEvent("test"));
             await Task.Delay(1000);
             _storage.Verify(o => o.Write(StorageConstants.Events, It.IsAny<string>()), Times.Never);
 
             // make sure uploadChannel is stopped
-            _eventPipeline.Flush();
+            eventPipeline.Flush();
             await Task.Delay(1000);
             _storage.Verify(o => o.Rollover(), Times.Never);
             _storage.Verify(o => o.Read(StorageConstants.Events), Times.Never);
@@ -123,12 +129,14 @@ public async void TestStop()
             _storage.Verify(o => o.RemoveFile(_file), Times.Never);
         }
 
-        [Fact]
-        public async Task TestFlushCausedByOverflow()
+        [Theory]
+        [MemberData(nameof(GetPipelineProvider))]
+        public async Task TestFlushCausedByOverflow(IEventPipelineProvider provider)
         {
-            _eventPipeline.Start();
-            _eventPipeline.Put(new ScreenEvent("event 1"));
-            _eventPipeline.Put(new ScreenEvent("event 2"));
+            IEventPipeline eventPipeline = provider.Create(_analytics, "key");
+            eventPipeline.Start();
+            eventPipeline.Put(new ScreenEvent("event 1"));
+            eventPipeline.Put(new ScreenEvent("event 2"));
 
             await Task.Delay(1000);
 
@@ -138,9 +146,11 @@ public async Task TestFlushCausedByOverflow()
             _storage.Verify(o => o.RemoveFile(_file), Times.Exactly(1));
         }
 
-        [Fact]
-        public async Task TestPeriodicalFlush()
+        [Theory]
+        [MemberData(nameof(GetPipelineProvider))]
+        public async Task TestPeriodicalFlush(IEventPipelineProvider provider)
         {
+            IEventPipeline eventPipeline = provider.Create(_analytics, "key");
             foreach (IFlushPolicy policy in _analytics.Configuration.FlushPolicies)
             {
                 if (policy is FrequencyFlushPolicy)
@@ -151,21 +161,21 @@ public async Task TestPeriodicalFlush()
             _analytics.AddFlushPolicy(new FrequencyFlushPolicy(1000L));
 
             // since we set autoAddSegmentDestination = false, we need to manually add it to analytics.
-            // we need a mocked SegmentDestination so we can redirect Flush call to this _eventPipeline.
+            // we need a mocked SegmentDestination so we can redirect Flush call to this eventPipeline.
             var segmentDestination = new Mock<SegmentDestination>();
-            segmentDestination.Setup(o => o.Flush()).Callback(() => _eventPipeline.Flush());
+            segmentDestination.Setup(o => o.Flush()).Callback(() => eventPipeline.Flush());
             segmentDestination.Setup(o => o.Analytics).Returns(_analytics);
             _analytics.Add(segmentDestination.Object);
 
-            _eventPipeline = new EventPipeline(
+            eventPipeline = new EventPipeline(
                 _analytics,
                 logTag: "key",
                 apiKey: _analytics.Configuration.WriteKey,
                 flushPolicies: _analytics.Configuration.FlushPolicies,
                 apiHost: _analytics.Configuration.ApiHost
             );
-            _eventPipeline.Start();
-            _eventPipeline.Put(new ScreenEvent("test"));
+            eventPipeline.Start();
+            eventPipeline.Put(new ScreenEvent("test"));
 
             await Task.Delay(2050);
 
@@ -175,16 +185,18 @@ public async Task TestPeriodicalFlush()
             _storage.Verify(o => o.RemoveFile(_file), Times.Exactly(2));
         }
 
-        [Fact]
-        public async Task TestFlushInterruptedWhenNoFileExist()
+        [Theory]
+        [MemberData(nameof(GetPipelineProvider))]
+        public async Task TestFlushInterruptedWhenNoFileExist(IEventPipelineProvider provider)
         {
+            IEventPipeline eventPipeline = provider.Create(_analytics, "key");
             // make sure the file does not exist
             _storage
                 .Setup(o => o.ReadAsBytes(It.IsAny<string>()))
                 .Returns((byte[])null);
 
-            _eventPipeline.Start();
-            _eventPipeline.Flush();
+            eventPipeline.Start();
+            eventPipeline.Flush();
 
             await Task.Delay(1000);
 

From 300bcfb8627130dbe5ac519111d5bf14f5b40f6c Mon Sep 17 00:00:00 2001
From: Michael Grosse Huelsewiesche <mihuelsewiesche@twilio.com>
Date: Tue, 30 Apr 2024 12:23:10 -0400
Subject: [PATCH 4/4] Changing to SemaphoreSlim which works well cross
 platform, supports cancellation tokens, and is not imitating.

---
 .../Analytics/Utilities/SyncEventPipeline.cs        | 13 ++++++++-----
 .../Utilities/SyncEventPipelineProvider.cs          |  7 ++++---
 2 files changed, 12 insertions(+), 8 deletions(-)

diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs
index f8b18ee..a52565b 100644
--- a/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipeline.cs
@@ -11,9 +11,9 @@ namespace Segment.Analytics.Utilities
     internal sealed class FlushEvent : RawEvent
     {
         public override string Type => "flush";
-        public readonly Semaphore _semaphore;
+        public readonly SemaphoreSlim _semaphore;
 
-        internal FlushEvent(Semaphore semaphore)
+        internal FlushEvent(SemaphoreSlim semaphore)
         {
             _semaphore = semaphore;
         }
@@ -41,6 +41,7 @@ public class SyncEventPipeline: IEventPipeline
         public bool Running { get; private set; }
 
         internal int _flushTimeout = -1;
+        internal CancellationToken _flushCancellationToken = CancellationToken.None;
 
         public SyncEventPipeline(
             Analytics analytics,
@@ -48,7 +49,8 @@ public SyncEventPipeline(
             string apiKey,
             IList<IFlushPolicy> flushPolicies,
             string apiHost = HTTPClient.DefaultAPIHost,
-            int flushTimeout = -1)
+            int flushTimeout = -1,
+            CancellationToken? flushCancellationToken = null)
         {
             _analytics = analytics;
             _logTag = logTag;
@@ -62,14 +64,15 @@ public SyncEventPipeline(
             _storage = analytics.Storage;
             Running = false;
             _flushTimeout = flushTimeout;
+            _flushCancellationToken = flushCancellationToken ?? CancellationToken.None;
         }
 
         public void Put(RawEvent @event) => _writeChannel.Send(@event);
 
         public void Flush() {
-            FlushEvent flushEvent = new FlushEvent(new Semaphore(1,1));
+            FlushEvent flushEvent = new FlushEvent(new SemaphoreSlim(1,1));
             _writeChannel.Send(flushEvent);
-            flushEvent._semaphore.WaitOne(_flushTimeout);
+            flushEvent._semaphore.Wait(_flushTimeout, _flushCancellationToken);
         } 
 
         public void Start()
diff --git a/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs
index 17f5373..5794677 100644
--- a/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs
+++ b/Analytics-CSharp/Segment/Analytics/Utilities/SyncEventPipelineProvider.cs
@@ -4,11 +4,11 @@ namespace Segment.Analytics.Utilities
 {
     public class SyncEventPipelineProvider: IEventPipelineProvider
     {
-        internal int _flushTimeout = 5000;
+        internal int _flushTimeout = -1;
         internal CancellationToken? _flushCancellationToken = null;
 
         public SyncEventPipelineProvider(
-            int flushTimeout = 5000,
+            int flushTimeout = -1,
             CancellationToken? flushCancellationToken = null)
         {
             _flushTimeout = flushTimeout;
@@ -21,7 +21,8 @@ public IEventPipeline Create(Analytics analytics, string key)
                     analytics.Configuration.WriteKey,
                     analytics.Configuration.FlushPolicies,
                     analytics.Configuration.ApiHost,
-                    _flushTimeout);
+                    _flushTimeout,
+                    _flushCancellationToken);
         }
     }
 }
\ No newline at end of file
