Add FileStream and NetworkStream Receiver Abstractions (#103)

* Support NetworkStreamNmeaReceiver and FileStreamNmeaReceiver for online / offline demos.

* Remove unused reference

* Switch back to networkstream

Solutions/Ais.Net.Receiver.Host.Console/Program.cs

@@ -35,7 +35,15 @@ public static async Task Main(string[] args)
 
             IStorageClient storageClient = new AzureAppendBlobStorageClient(storageConfig);
 
-            var receiverHost = new ReceiverHost(aisConfig);
+            INmeaReceiver receiver = new NetworkStreamNmeaReceiver(
+                aisConfig.Host,
+                aisConfig.Port,
+                aisConfig.RetryPeriodicity,
+                aisConfig.RetryAttempts);
+
+            // INmeaReceiver receiver = new FileStreamNmeaReceiver(@"PATH-TO-RECORDING.nm4");
+
+            var receiverHost = new ReceiverHost(receiver);
 
             // Decode teh sentences into messages, and group by the vessel by Id
             IObservable<IGroupedObservable<uint, IAisMessage>> byVessel = receiverHost.Messages.GroupBy(m => m.Mmsi);

Solutions/Ais.Net.Receiver/Ais.Net.Receiver.csproj

@@ -21,7 +21,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Corvus.Retry" Version="1.0.1" />
+    <PackageReference Include="Corvus.Retry" Version="1.0.2" />
     <PackageReference Include="Endjin.RecommendedPractices" Version="1.2.0">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

Solutions/Ais.Net.Receiver/Ais/Net/Receiver/Receiver/FileStreamNmeaReceiver.cs

@@ -0,0 +1,43 @@
+// <copyright file="NmeaReceiver.cs" company="Endjin Limited">
+// Copyright (c) Endjin Limited. All rights reserved.
+// </copyright>
+
+namespace Ais.Net.Receiver.Receiver
+{
+    using System;
+    using System.Collections.Generic;
+    using System.IO;
+    using System.Runtime.CompilerServices;
+    using System.Threading;
+
+    public class FileStreamNmeaReceiver : INmeaReceiver
+    {
+        private readonly string path;
+
+        public FileStreamNmeaReceiver(string path)
+        {
+            this.path = path;
+        }
+
+        public int RetryAttemptLimit { get; }
+
+        public TimeSpan RetryPeriodicity { get; }
+
+        public async IAsyncEnumerable<string> GetAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
+        {
+            using var sr = new StreamReader(this.path);
+
+            while (sr.Peek() >= 0)
+            {
+                if (cancellationToken.IsCancellationRequested)
+                {
+                    break;
+                }
+
+                string? line = await sr.ReadLineAsync().ConfigureAwait(false);
+
+                if (line is not null) { yield return line; }
+            }
+        }
+    }
+}

Solutions/Ais.Net.Receiver/Ais/Net/Receiver/Receiver/INmeaReceiver.cs

@@ -0,0 +1,14 @@
+// <copyright file="NmeaReceiver.cs" company="Endjin Limited">
+// Copyright (c) Endjin Limited. All rights reserved.
+// </copyright>
+
+namespace Ais.Net.Receiver.Receiver
+{
+    using System.Collections.Generic;
+    using System.Threading;
+
+    public interface INmeaReceiver
+    {
+        IAsyncEnumerable<string> GetAsync(CancellationToken cancellationToken = default);
+    }
+}

Solutions/Ais.Net.Receiver/Ais/Net/Receiver/Receiver/NmeaReceiver.cs → Solutions/Ais.Net.Receiver/Ais/Net/Receiver/Receiver/NetworkStreamNmeaReceiver.cs

@@ -12,11 +12,11 @@ namespace Ais.Net.Receiver.Receiver
     using System.Threading;
     using System.Threading.Tasks;
 
-    public class NmeaReceiver
+    public class NetworkStreamNmeaReceiver : INmeaReceiver
     {
         private readonly TcpClient tcpClient = new();
 
-        public NmeaReceiver(string host, int port, TimeSpan? retryPeriodicity, int retryAttemptLimit = 100)
+        public NetworkStreamNmeaReceiver(string host, int port, TimeSpan? retryPeriodicity, int retryAttemptLimit = 100)
         {
             this.Host = host;
             this.Port = port;

Solutions/Ais.Net.Receiver/Ais/Net/Receiver/Receiver/ReceiverHost.cs

@@ -13,7 +13,6 @@ namespace Ais.Net.Receiver.Receiver
     using System.Threading.Tasks;
 
     using Ais.Net.Models.Abstractions;
-    using Ais.Net.Receiver.Configuration;
     using Ais.Net.Receiver.Parser;
 
     using Corvus.Retry;
@@ -22,19 +21,13 @@ namespace Ais.Net.Receiver.Receiver
 
     public class ReceiverHost
     {
-        private readonly AisConfig configuration;
-        private readonly NmeaReceiver receiver;
+        private readonly INmeaReceiver receiver;
         private readonly Subject<string> sentences = new();
         private readonly Subject<IAisMessage> messages = new();
 
-        public ReceiverHost(AisConfig configuration)
+        public ReceiverHost(INmeaReceiver receiver)
         {
-            this.configuration = configuration;
-            this.receiver = new NmeaReceiver(
-                this.configuration.Host,
-                this.configuration.Port,
-                this.configuration.RetryPeriodicity,
-                this.configuration.RetryAttempts);
+            this.receiver = receiver;
         }
 
         public IObservable<string> Sentences => this.sentences;