Merge pull request #11 from sceneskope/dev

Added base reading receiver
sceneskope · Aug 3, 2018 · c063fec · c063fec
2 parents 8f63383 + ff67ea8
commit c063fec
Show file tree

Hide file tree

Showing 4 changed files with 65 additions and 55 deletions.
diff --git a/directory.build.props b/directory.build.props
@@ -9,7 +9,7 @@
     <PackageLicenseUrl>https://opensource.org/licenses/MIT</PackageLicenseUrl>
     <PackageRequireLicenseAcceptance>true</PackageRequireLicenseAcceptance>
     <LangVersion>latest</LangVersion>
-    <VersionPrefix>5.1.1</VersionPrefix>
+    <VersionPrefix>5.2.0</VersionPrefix>
     <DebugType>embedded</DebugType>
   </PropertyGroup>
   <PropertyGroup Condition="'$(IsTestProject)' != 'true' ">

diff --git a/src/SceneSkope.ServiceFabric.EventHubs/BaseReadingReceiver.cs b/src/SceneSkope.ServiceFabric.EventHubs/BaseReadingReceiver.cs
@@ -0,0 +1,52 @@
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Microsoft.Azure.EventHubs;
+using Microsoft.ServiceFabric.Data;
+using Microsoft.ServiceFabric.Data.Collections;
+using Serilog;
+using ServiceFabric.Utilities;
+
+namespace SceneSkope.ServiceFabric.EventHubs
+{
+    public abstract class BaseReadingReceiver : IReadingReceiver
+    {
+        public Func<Exception, bool> TransientExceptionChecker { get; }
+        public ServiceFabricRetryHandler RetryHandler { get; }
+        public ILogger Log { get; }
+        public IReliableStateManager StateManager { get; }
+
+        protected readonly IReliableDictionary<string, string> _offsets;
+        protected readonly string _partition;
+
+        public PartitionReceiver Receiver { get; }
+
+        protected BaseReadingReceiver(ILogger log,
+            IReliableStateManager stateManager,
+            PartitionReceiver receiver,
+            IReliableDictionary<string, string> offsets, string partition,
+            ServiceFabricRetryHandler retryHandler,
+            Func<Exception, bool> transientExceptionChecker = null)
+        {
+            RetryHandler = retryHandler;
+            Log = log;
+            StateManager = stateManager;
+            Receiver = receiver;
+            _offsets = offsets;
+            _partition = partition;
+            TransientExceptionChecker = transientExceptionChecker;
+        }
+
+        public virtual Task InitialiseAsync() => Task.CompletedTask;
+
+        protected Task SaveOffsetAsync(ITransaction tx, string latestOffset) => _offsets.SetAsync(tx, _partition, latestOffset);
+
+        public virtual void Dispose()
+        {
+        }
+
+        public Task ProcessEventsAsync(IEnumerable<EventData> events) => ProcessEventsAsync((IReadOnlyList<EventData>)events);
+
+        protected abstract Task ProcessEventsAsync(IReadOnlyList<EventData> events);
+    }
+}
diff --git a/src/SceneSkope.ServiceFabric.EventHubs/BatchedReadingReceiver.cs b/src/SceneSkope.ServiceFabric.EventHubs/BatchedReadingReceiver.cs
@@ -10,39 +10,23 @@
 
 namespace SceneSkope.ServiceFabric.EventHubs
 {
-    public abstract class BatchedReadingReceiver : IReadingReceiver
+    public abstract class BatchedReadingReceiver : BaseReadingReceiver
     {
-        public Func<Exception, bool> TransientExceptionChecker { get; }
-        public ServiceFabricRetryHandler RetryHandler { get; }
-        public ILogger Log { get; }
-        public IReliableStateManager StateManager { get; }
-
-        private readonly IReliableDictionary<string, string> _offsets;
-        protected readonly string _partition;
-
-        public PartitionReceiver Receiver { get; }
 
         protected BatchedReadingReceiver(ILogger log,
             IReliableStateManager stateManager,
             PartitionReceiver receiver,
-            IReliableDictionary<string, string> offsets, string partition,
+            IReliableDictionary<string, string> offsets, 
+            string partition,
             ServiceFabricRetryHandler retryHandler,
-            Func<Exception, bool> transientExceptionChecker = null)
+            Func<Exception, bool> transientExceptionChecker = null) : 
+            base(log, stateManager, receiver, offsets, partition, retryHandler, transientExceptionChecker)
         {
-            RetryHandler = retryHandler;
-            Log = log;
-            StateManager = stateManager;
-            Receiver = receiver;
-            _offsets = offsets;
-            _partition = partition;
-            TransientExceptionChecker = transientExceptionChecker;
         }
 
-        public virtual Task InitialiseAsync() => Task.CompletedTask;
-
         protected abstract Task ProcessEventAsync(ITransaction tx, EventData @event, CancellationToken serviceCancellationToken);
 
-        public async Task ProcessEventsAsync(IEnumerable<EventData> events)
+        protected override async Task ProcessEventsAsync(IReadOnlyList<EventData> events)
         {
             await BeforeProcessEventsAsync(RetryHandler.ServiceCancellationToken).ConfigureAwait(false);
             await RetryHandler.HandleAsync(async cancel =>
@@ -69,8 +53,5 @@ await RetryHandler.HandleAsync(async cancel =>
         public virtual Task BeforeProcessEventsAsync(CancellationToken serviceCancellationToken) => Task.CompletedTask;
         public virtual Task AfterProcessEventsAsync(CancellationToken serviceCancellationToken) => Task.CompletedTask;
 
-        public virtual void Dispose()
-        {
-        }
     }
 }
diff --git a/src/SceneSkope.ServiceFabric.EventHubs/SimpleReadingReceiver.cs b/src/SceneSkope.ServiceFabric.EventHubs/SimpleReadingReceiver.cs
@@ -9,39 +9,22 @@
 
 namespace SceneSkope.ServiceFabric.EventHubs
 {
-    public abstract class SimpleReadingReceiver : IReadingReceiver
+    public abstract class SimpleReadingReceiver : BaseReadingReceiver
     {
-        public Func<Exception, bool> TransientExceptionChecker { get; }
-        public ServiceFabricRetryHandler RetryHandler { get; }
-        public ILogger Log { get; }
-        public IReliableStateManager StateManager { get; }
-
-        private readonly IReliableDictionary<string, string> _offsets;
-        protected readonly string _partition;
-
-        public PartitionReceiver Receiver { get; }
 
         protected SimpleReadingReceiver(ILogger log,
             IReliableStateManager stateManager,
             PartitionReceiver receiver,
-            IReliableDictionary<string, string> offsets, string partition,
+            IReliableDictionary<string, string> offsets, 
+            string partition,
             ServiceFabricRetryHandler retryHandler,
-            Func<Exception, bool> transientExceptionChecker = null)
+            Func<Exception, bool> transientExceptionChecker = null) :
+                base(log, stateManager, receiver, offsets, partition, retryHandler, transientExceptionChecker)
         {
-            RetryHandler = retryHandler;
-            Log = log;
-            StateManager = stateManager;
-            Receiver = receiver;
-            _offsets = offsets;
-            _partition = partition;
-            TransientExceptionChecker = transientExceptionChecker;
         }
-
-        public virtual Task InitialiseAsync() => Task.CompletedTask;
-
         protected abstract Task ProcessEventAsync(EventData @event);
 
-        public async Task ProcessEventsAsync(IEnumerable<EventData> events)
+        protected override async Task ProcessEventsAsync(IReadOnlyList<EventData> events)
         {
             string latestOffset = null;
             foreach (var @event in events)
@@ -53,11 +36,5 @@ public async Task ProcessEventsAsync(IEnumerable<EventData> events)
         }
 
         protected virtual Task OnAllEventsProcessedAsync(string latestOffset) => Task.CompletedTask;
-
-        protected Task SaveOffsetAsync(ITransaction tx, string latestOffset) => _offsets.SetAsync(tx, _partition, latestOffset);
-
-        public virtual void Dispose()
-        {
-        }
     }
 }