A simple .NET AIS Receiver for capturing the Norwegian Coastal Administration AIS network data (available under Norwegian license for public data (NLOD)) and persisting in Microsoft Azure Blob Storage. Sponsored by endjin.
License: Apache License 2.0
![dependabot[bot] avatar](https://avatars.githubusercontent.com/in/29110?v=4 "dependabot[bot]")
![dependjinbot[bot] avatar](https://avatars.githubusercontent.com/in/80870?v=4 "dependjinbot[bot]")
Unhandled Exception: Microsoft.Azure.Storage.StorageException: The append position condition specified was not met.
at Microsoft.Azure.Storage.Blob.BlobWriteStreamBase.ThrowLastExceptionIfExists()
at Microsoft.Azure.Storage.Blob.BlobWriteStream.FlushAsync(CancellationToken token)
at Microsoft.Azure.Storage.Blob.BlobWriteStream.CommitAsync()
at Microsoft.Azure.Storage.Blob.BlobWriteStream.Dispose(Boolean disposing)
at System.IO.Stream.Close()
at Microsoft.Azure.Storage.Blob.CloudAppendBlob.UploadFromStreamHelper(Stream source, Nullable`1 length, Boolean createNew, AccessCondition accessCondition, BlobRequestOptions options, OperationContext operationContext)
at Microsoft.Azure.Storage.Blob.CloudAppendBlob.AppendFromStream(Stream source, AccessCondition accessCondition, BlobRequestOptions options, OperationContext operationContext)
at Microsoft.Azure.Storage.Blob.CloudAppendBlob.AppendFromByteArray(Byte[] buffer, Int32 index, Int32 count, AccessCondition accessCondition, BlobRequestOptions options, OperationContext operationContext)
at Microsoft.Azure.Storage.Blob.CloudAppendBlob.AppendText(String content, Encoding encoding, AccessCondition accessCondition, BlobRequestOptions options, OperationContext operationContext)
at Endjin.Ais.Receiver.StorageClient.AppendMessages(IList`1 messages) in C:\Ais.Net.Receiver\Solutions\Ais.Net.Receiver\StorageClient.cs:line 43
at System.Reactive.AnonymousSafeObserver`1.OnNext(T value) in D:\a\1\s\Rx.NET\Source\src\System.Reactive\AnonymousSafeObserver.cs:line 44
at System.Reactive.Subjects.Subject`1.OnNext(T value) in D:\a\1\s\Rx.NET\Source\src\System.Reactive\Subjects\Subject.cs:line 148
at Endjin.Ais.Receiver.NmeaReceiver.RecieveAsync() in C:\Ais.Net.Receiver\Solutions\Ais.Net.Receiver\NmeaReceiver.cs:line 62
at Endjin.Ais.Receiver.Program.Main(String[] args) in C:\Ais.Net.Receiver\Solutions\Ais.Net.Receiver\Program.cs:line 31
at Endjin.Ais.Receiver.Program.<Main>(String[] args)Bad AIS messages are a fact of life. The NmeaLineParser constructor detects various clear indications of bad data and throws an ArgumentException. The ReceiverHost does not handle these exceptions, causing it to crash if a bad message is received.
We should handle ArgumentException from this line:
We should consider adding a new IObservable<T> property reporting these bad lines, so that applications can still do something with them. (E.g., if we're recording all incoming messages, it may be useful to include incomplete/corrupt ones, because manual inspection might be able to infer something from the data.)
The library currently in use has now been marked as deprecated. Microsoft wants everyone to move over to the new Azure.* libraries.
Message type values are always in the range 0-47 inclusive, because there are only 6 bits available to hold the message type, and by definition they are non-negative.
While all of the various message parsers reflect this by defining a property of the form public uint MessageType => ..., the NmeaPayloadParser.PeekMessageType method inexplicably returns an int.
I can't think of any good reason why I would have done that, so I believe this was a simple mistake.
Since we're still in v0.x versioning, it's not technically a semver violation to fix this.
Unhandled Exception: System.Net.Internals.SocketExceptionFactory+ExtendedSocketException: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond 153.44.253.27:5631
at System.Net.Sockets.Socket.EndConnect(IAsyncResult asyncResult)
at System.Net.Sockets.TcpClient.EndConnect(IAsyncResult asyncResult)
at System.Net.Sockets.TcpClient.<>c.<ConnectAsync>b__28_1(IAsyncResult asyncResult)
at System.Threading.Tasks.TaskFactory`1.FromAsyncCoreLogic(IAsyncResult iar, Func`2 endFunction, Action`1 endAction, Task`1 promise, Boolean requiresSynchronization)
--- End of stack trace from previous location where exception was thrown ---
at Endjin.Ais.Receiver.NmeaReceiver.InitaliseAsync() in C:\Ais.Net.Receiver\Solutions\Ais.Net.Receiver\NmeaReceiver.cs:line 48
at Endjin.Ais.Receiver.Program.Main(String[] args) in C:\Ais.Net.Receiver\Solutions\Ais.Net.Receiver\Program.cs:line 33
at Endjin.Ais.Receiver.Program.<Main>(String[] args)Once #141 is implemented, we need to go on to decide what the correct behaviour is for the app that persists messages received via the NetworkStreamNmeaReceiver.
Should it record the lines anyway? Should it somehow include information in its recorded output saying that the line is bad? Should it include information from the exception? Should it record bad messages separately?
While there are many different AIS message types, the features in IAisMessage are available in all message types. Although this fact is reflected in the hierarchy of concrete message types (all the various record types representing specific message types such as AisMessageType1Through3 and AisMessageType5 derive from AisMessageBase which implements IAisMessage) it is not reflected in the interfaces themselves.
This leads to the following slightly annoying situation. Suppose I have some source of IAisMessages:
IObservable<IAisMessage> messages = GetAisMessageSource()and suppose I filter this down to messages that provide navigation information:
IObservable<IVesselNavigation> navigationMessages =
messages.OfType<IVesselNavigation>();the resulting navigationMessages makes it awkward to get hold of universal AIS features such as the MMSI.
navigationMessages .Subscribe(m => Console.WriteLine(
$"Ship {((IAisMessage)m).Mmsi} status is {m.NavigationStatus}");This code has had to cast the message to IAisMessage in order to use its Mmsi property.
The MMSI is a universal feature of AIS messages. (That's why it's safe for that code to cast all messages to IAisMessage.) This should be reflected in the various interface definitions. We should never have to cast back to an IAisMessage—the availability of those features should be reflected by the interface type hierarchy.
(Other fixed relationships like this should also be represented. For example, we define IAisMessageType1to3. Messages of this kind will invariably offer IRaimFlag, IRepeatIndicator, and IVesselNavigation as well as IAisMessage. This is not currently evident from the IAisMessageType1to3 definition, but it should be.)
When the AIS Receiver was running for over 7 days a fatal exception occurred while attempting to write a batch of messages to blob storage:
Unhandled Exception: Microsoft.Azure.Storage.StorageException: Server failed to authenticate the request. Make sure the value of Authorization header is formed correctly including the signature.
at Microsoft.Azure.Storage.Core.Executor.Executor.ExecuteAsync[T](RESTCommand 1 cmd, IRetryPolicy policy, OperationContext operationContext, CancellationToken token)
at Microsoft.Azure.Storage.Core.Executor.Executor.<>cDisplayClass0_0 1.<ExecuteSync>b0()
at Microsoft.Azure.Storage.Core.Uti1.CommonUti1ity.RunWithoutSynchronizationContext[T](Func'1 actionToRun) at Microsoft.Azure.Storage.Core.Executor.Executor.ExecuteSync[T](RESTCommand 1 cmd, IRetryPolicy policy, OperationCon text operationContext)
at Endjin.Ais.Receiver.Program.GetAppendBlob() in C:\_Proj ects\OSS\Endjin.Ais.Receiver\Endjin.Ais.Receiver\Program.cs :line 98
at Endiin.Ais.Receiyer.Program.OnMessageReceived(lList 1 messages) in C:\Endjin.Ais.Receiver\Program.cs:line 83
at System.Reactive.AnonymousSafeObserver 1.OnNext(T value) in D:\a\l\s\Rx.NET\Source\src\System.Reactive\AnonymousSafeObserver.csrline 44
at System.Reactive.Subjects.Subject 1.OnNext(T value) in D:\a\l\s\Rx.NET\Source\src\System.Reactive\Subjects\Subject.cs:line 148
at Endjin.Ais.Receiver.NmeaReceiver.RecieveAsync() in C:\Endjin.Ais.Receiver\NmeaReceiver.csrline 59
at Endjin.Ais.Receiver.Program.LongRunningOperationAsync() in C:\Endjin.Ais.Receiver\Program.cs:1ine 74
at Endjin.Ais.Receiyer.Program.LonqRunningOperationWithCancel1ationTokenAsync(Cancel1ationToken cancel1ationToken) in C:\Endjin.Ais.Receiver\Program.cs:1ine 61
at Enajin.Ais.Receiver.Program.Main(String[] args) in C:\Endjin.Ais.Receiver\Progra m.cs:line 33
at Endjin.Ais.Receiver.Program.<Main>(String[] args)The AIS Transmitter sometimes hangs and needs a reboot, when this happens the AIS Receiver crashes with the following exception.
Unhandled Exception: System.Net.Internals.SocketExceptionFactory+ExtendedSocketException: No connection could be made because the target machine actively refused it 153.44.253.27:5631 at System.Net.Sockets.Socket.EndConnect(IAsyncResult asyncResult) at System.Net.Sockets.TcpClient.EndConnect(IAsyncResult asyncResult) at System.Net.Sockets.TcpClient.<>c.<ConnectAsync>b 28_l(IAsyncResult asyncResult)
at System.Threading.Tasks.TaskFactory'1.FromAsyncCoreLogic(IAsyncResult iar, Func'2 endFunction, Action'1 endAction, Task'1 promise. Boolean requiresSynchronization)
— End of stack trace from previous location where exception was thrown —
at Endjin.Ais.Receiver.NmeaReceiver.InitaliseAsync() in C:\Ais.Net.Receiver\NmeaReceiver.cs:line 48 at Endjin.Ais.Receiver.Program.Main(String[] args) in C:\Ais.Net.Receiver\Program.cs:line 33 at Endjin.Ais.Receiver.Program.<Main>(String[] args)The readme of the Ais.Net project claims support for message type 27 but as far as I can tell there is not any support for this meesage type, would be great to add it.
We've been having issues where the network stream keeps stopping occasionally for a few seconds until it resumes. There are no logged error messages, and when it resumes the stream is as fast as before, so we're losing a lot of data. Eventually, after several minutes, it results in Corvus failing with the following error message:
System.ArgumentOutOfRangeException: The value needs to translate in milliseconds to -1 (signifying an infinite timeout), 0, or a positive integer less than or equal to the maximum allowed timer duration. (Parameter 'delay')
at System.Threading.Tasks.Task.ValidateTimeout(TimeSpan timeout, ExceptionArgument argument)
at System.Threading.Tasks.Task.Delay(TimeSpan delay)
at Corvus.Retry.Async.SleepService.DefaultSleepService.SleepAsync(TimeSpan timeSpan)
at Corvus.Retry.Retriable.RetryAsync(Func`1 asyncFunc, CancellationToken cancellationToken, IRetryStrategy strategy, IRetryPolicy policy, Boolean continueOnCapturedContext)
at AisLog.Services.AisLogService.ExecuteAsync(CancellationToken cancellationToken) in /source/Services/AisLogService.cs:line 102
at Microsoft.Extensions.Hosting.Internal.Host.TryExecuteBackgroundServiceAsync(BackgroundService backgroundService)
This started happening a few weeks ago on our local development setups, and just recently it's also affecting our production server. CPU load is rarely high on either setup (<1%).
I've conducted some tests of running the receiver in an isolated environment, counting the number of messages received in a minute.
using Ais.Net.Receiver.Receiver;
var receiver = new NetworkStreamNmeaReceiver("153.44.253.27", 5631, TimeSpan.FromSeconds(0.5));
var receiverHost = new ReceiverHost(receiver);
var count = 0;
receiverHost.Sentences.Subscribe(_ => count++);
//receiverHost.Messages.Subscribe(_ => {});
await Task.WhenAny(receiverHost.StartAsync(), Task.Delay(TimeSpan.FromMinutes(1)));
Console.WriteLine(count);Here are results of 5 runs from my laptop:
$ dotnet run -> 2473
$ dotnet run -> 2410
$ dotnet run -> 2402
$ dotnet run -> 2400
$ dotnet run -> 2401
Subscribing to ReceiverHost.Sentences seemingly works fine, as every test I've run has resulted in between 2400-2500 messages.
However, uncommenting receiverHost.Messages.Subscribe(_ => {}); thus also parsing the messages using a subscription, I've been getting mixed results on the sentence count. Another 5 runs of this new code on my laptop:
$ dotnet run -> 743
$ dotnet run -> 499
$ dotnet run -> 935
$ dotnet run -> 837
$ dotnet run -> 373
I've also tried only subscribing to messages and counting those which results in similar numbers. Running tests with something like receiverHost.Sentences.Subscribe(Console.WriteLine); I can also inspect that subscribing to Sentences results in a steady stream of messages, whereas when subscribing to Messages the stream pauses multiple times.
The reason I've been specifying laptop is that it seems to run fine (at the moment) on my desktop machine, which understandably is more powerful.
Laptop: Dell XPS 13 9310, Linux Ubuntu 22.04.1 LTS, Intel Core i7-1185G7, 16 GB RAM
Desktop: Windows 10, AMD Ryzen 5600X, 32 GB RAM
Production server: Virtual LXC Server, Linux Ubuntu 20.04 LTS, 2 vCPUs (unsure but might be Intel Core i7-3770), 2 GB RAM
What could be the cause of this? Consiering Ais.Net reportedly can handle more than a million sentences per minute, I'd assume this is a problem in the receiver.
public Task StartAsync(CancellationToken cancellationToken = default)
{
return Retriable.RetryAsync(() =>
this.StartAsyncInternal(cancellationToken),
cancellationToken,
new Backoff(maxTries: 100, deltaBackoff: TimeSpan.FromSeconds(5)),
new AnyExceptionPolicy(),
false);
}retry time 18 = -24.20:31:22.6480000
Backoff is an exponentially increasing retry delay, which is not very suitable here and will lead to long retry waiting
Must be non-negative and less than the size of the collection. Arg_ParamName_Name
at System.ThrowHelper.ThrowArgumentOutOfRange_IndexException()
at System.Collections.Generic.List`1.get_Item(Int32 index)
at Ais.Net.Receiver.Host.Console.ReceiverHostExtensions.<>c.<GetStreamStatistics>b__0_1(IList`1 window) in \Ais.Net.Receiver\Solutions\Ais.Net.Receiver.Host.Console\Ais\Net\Receiver\Host\Console\ReceiverHostExtensions.cs:line 34
at System.Reactive.Linq.ObservableImpl.Select`2.Selector._.OnNext(TSource value) in /_/Rx.NET/Source/src/System.Reactive/Linq/Observable/Select.cs:line 39
--- End of stack trace from previous location ---
at System.Reactive.AnonymousSafeObserver`1.OnError(Exception error) in /_/Rx.NET/Source/src/System.Reactive/AnonymousSafeObserver.cs:line 62
at System.Reactive.Sink`2.OnError(Exception error) in /_/Rx.NET/Source/src/System.Reactive/Internal/Sink.cs:line 94
at System.Reactive.Sink`1.ForwardOnError(Exception error) in /_/Rx.NET/Source/src/System.Reactive/Internal/Sink.cs:line 61
at System.Reactive.Linq.ObservableImpl.Select`2.Selector._.OnNext(TSource value) in /_/Rx.NET/Source/src/System.Reactive/Linq/Observable/Select.cs:line 41
at System.Reactive.Linq.ObservableImpl.Buffer`1.TimeHopping._.Tick() in /_/Rx.NET/Source/src/System.Reactive/Linq/Observable/Buffer.cs:line 468
at System.Reactive.Concurrency.Scheduler.<>c__67`1.<SchedulePeriodic>b__67_0(ValueTuple`2 t) in /_/Rx.NET/Source/src/System.Reactive/Concurrency/Scheduler.Services.Emulation.cs:line 79
at System.Reactive.Concurrency.DefaultScheduler.PeriodicallyScheduledWorkItem`1.<>c.<Tick>b__5_0(PeriodicallyScheduledWorkItem`1 closureWorkItem) in /_/Rx.NET/Source/src/System.Reactive/Concurrency/DefaultScheduler.cs:line 127
at System.Reactive.Concurrency.AsyncLock.Wait(Object state, Delegate delegate, Action`2 action) in /_/Rx.NET/Source/src/System.Reactive/Concurrency/AsyncLock.cs:line 93
at System.Threading.TimerQueueTimer.<>c.<.cctor>b__27_0(Object state)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state)
--- End of stack trace from previous location ---
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state)
at System.Threading.TimerQueueTimer.CallCallback(Boolean isThreadPool)
at System.Threading.TimerQueueTimer.Fire(Boolean isThreadPool)
at System.Threading.TimerQueue.FireNextTimers()
at System.Threading.TimerQueue.AppDomainTimerCallback(Int32 id)
ReceiverHostExtensionsGetStreamStatistics line 34:
return runningCounts.Buffer(period)
.Select(window => (window[0], window[^1])How about adding support for Message types 4 and 21? These fixed location messages may not be as interesting as mobile, but are important for shipboard navigation. Unclear to me if this app/library was mostly designed as a land-based application. Also I'm using them to study VHF propagation changes.
Currently I'm using python and pyais but my python expertise is not great, and python appears to have less mature support for threading, multiprocessing etc.
The NmeaAisLongRangeAisBroadcastParser constructor docs say that it creates a new NmeaAisPositionReportClassAParser!
The Receiver encountered a transient error whereby Azure Storage could not be reached. The following exception was thrown:
Unhandled Exception: Microsoft.Azure.Storage.StorageException: No such host is known —> System.Net.Http.HttpRequestException: No such host is known —> System.Net.Sockets.SocketException: No such host is known
at System.Net.Http.ConnectHelper.ConnectAsync(String host, Int32 port, CancellationToken cancellationToken)
End of inner exception stack trace
at System.Net.Http.ConnectHelper.ConnectAsync(String host, Int32 port, CancellationToken cancellationToken) at System.Threading.Tasks.ValueTask 1.get_Result()
at System.Net.Http.HttpConnectionPool.CreateConnectionAsync(HttpRequestMessage request, CancellationToken cancellationToken)
at System.Threading.Tasks.ValueTask 1.get_Result()
at System.Net.Http.HttpConnectionPool.WaitForCreatedConnectionAsync(ValueTask 1 creationTask) at System.Threading.Tasks.ValueTask 1.get_Result()
at System.Net.Http.HttpConnectionPool.SendWithRetryAsync(HttpRequestMessage request, Boolean doRequestAuth, Cancellat ionToken cancellationToken)
at System.Net.Http.RedirectHandler.SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) at System.Net.Http.HttpClient.FinishSendAsyncUnbuffered(Task`1 senaTask, HttpRequestMessage request, CancellationTokenSource cts, Boolean disposeCts)
at Microsoft.Azure.Storage.Core.Executor.Executor.ExecuteAsync[T](RESTCommand`1 cmd, IRetryPolicy policy, OperationContext operationContext, CancellationToken token)
End of inner exception stack trace
at Microsoft.Azure.Storage.Blob.BlobWriteStreamBase.ThrowLastExceptionlfExists() at Microsoft.Azure.Storage.Blob.BlobWriteStream.FlushAsync(Cancel1ationToken token) at Microsoft.Azure.Storage.Blob.BlobWriteStream.CommitAsync() at Microsoft.Azure.Storage.Blob.BlobWriteStream.Dispose Boolean disposing) at System`10.Stream.Close()
at Microsoft.Azure.Storage.Blob.CloudAppendBlob.UploadFromStreamHelper(Stream source, Nullable`1 length, Boolean createNew, AccessCondition accessCondition, BlobRequestOptions options, OperationContext operationContext)
at Microsoft.Azure.Storage.Blob.CloudAppendBlob.AppendFromStream(Stream source, AccessCondition accessCondition, Blob RequestOptions options, OperationContext operationContext)
at Microsoft.Azure.Storage.Blob.CloudAppendBlob.AppendFromByteArray(Byte[] buffer, Int32 index, Int32 count, AccessCondition accessCondition, BlobRequestOptions options, OperationContext operationContext)
at Microsoft.Azure.Storage.Blob.CloudAppendBlob.AppendText(String content, Encoding encoding, AccessCondition accessCondition, BlobRequestOptions options, OperationContext operationContext)
at Endjin.Ai s.Receiver.StorageClient.AppendMessages(IList`1 messages) in C:\Ais.Net.Receiver\Solutions\Ais.Net.Receiver\StorageClient.cs :1ine 43
at Endjin.Ais.Receiver.Program.OnMessageReceived(IList`1 messages) in C:\Ais.Net.Receiver\Solutions\Ais.Net.Receiver\Program.cs:line 51
at System.Reactive.AnonymousSafeObserver`1.OnNext(T value) in D:\a\l\s\Rx.NET\Source\src\System.Reactive\AnonymousSafeObserver.cs:line 44
at System.Reactive.Subjects.Subject`1.OnNext(T value) in D:\a\l\s\Rx.NET\Source\src\System.Reactive\Subjects\Subject.cs:line 148
at Endjin.Ais.Receiver.NmeaReceiver.RecieveAsync() in C:\Ais.Net.Receiver\Solutions\Ais.Net.Receiver\NmeaReceiver.cs:line 61
at Endiin.Ais.Receiver.Program.Main(String[] args) in C:\Ais.Net.Receiver\Solutions\Ais.Net.Receiver\Program.es:line 31
at Endjin.Ais.Receiver.Program.<Main>(String[] args)Consider refactoring the code to use TPL Dataflow and either using a BufferBlock to hold incoming messages while trying to re-establish a connection with Azure Storage, or write messages to a local repository, which are then transmitted, in order, when the connection is re-established.
Process terminated with the following stracktrace
at System.Reactive.Concurrency.ConcurrencyAbstractionLayerImpl.PeriodicTimer.<>c.<.ctor>b__2_0(Object this) in /_/Rx.NET/Source/src/System.Reactive/Concurrency/ConcurrencyAbstractionLayerImpl.cs:line 207
at System.Threading.TimerQueueTimer.<>c.<.cctor>b__27_0(Object state)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state)
--- End of stack trace from previous location ---
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state)
at System.Threading.TimerQueueTimer.Fire(Boolean isThreadPool)
at System.Threading.TimerQueue.FireNextTimers()
at System.Threading.TimerQueue.AppDomainTimerCallback(Int32 id)
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
Django
The Web framework for perfectionists with deadlines.
Laravel
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
Microsoft
Open source projects and samples from Microsoft.
Google
Google ❤️ Open Source for everyone.
Alibaba
Alibaba Open Source for everyone
D3
Data-Driven Documents codes.
Tencent
China tencent open source team.