Implement nullability in streams Buffer

src/core/Akka.Streams/Attributes.cs

@@ -7,6 +7,7 @@
 
 using System;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.Linq;
 using System.Text;
 using Akka.Event;
@@ -325,8 +326,11 @@ public IEnumerable<TAttr> GetAttributeList<TAttr>() where TAttr : IAttribute
         /// <typeparam name="TAttr">TBD</typeparam>
         /// <param name="defaultIfNotFound">TBD</param>
         /// <returns>TBD</returns>
-        public TAttr GetAttribute<TAttr>(TAttr defaultIfNotFound) where TAttr : class, IAttribute
+        #nullable enable
+        [return: NotNullIfNotNull("defaultIfNotFound")]
+        public TAttr? GetAttribute<TAttr>(TAttr? defaultIfNotFound) where TAttr : class, IAttribute
             => GetAttribute<TAttr>() ?? defaultIfNotFound;
+        #nullable restore
 
         /// <summary>
         /// Get the first (least specific) attribute of a given type or subtype thereof.

src/core/Akka.Streams/Implementation/ActorRefSourceActor.cs

@@ -10,6 +10,7 @@
 using Akka.Event;
 using Akka.Streams.Actors;
 
+#nullable enable
 namespace Akka.Streams.Implementation
 {
     /// <summary>
@@ -40,7 +41,7 @@ public static Props Props(int bufferSize, OverflowStrategy overflowStrategy, Act
         /// <summary>
         /// TBD
         /// </summary>
-        protected readonly IBuffer<T> Buffer;
+        protected readonly IBuffer<T>? Buffer;
 
         /// <summary>
         /// TBD
@@ -50,7 +51,6 @@ public static Props Props(int bufferSize, OverflowStrategy overflowStrategy, Act
         /// TBD
         /// </summary>
         public readonly OverflowStrategy OverflowStrategy;
-        private ILoggingAdapter _log;
 
         /// <summary>
         /// TBD
@@ -63,13 +63,13 @@ public ActorRefSourceActor(int bufferSize, OverflowStrategy overflowStrategy, in
         {
             BufferSize = bufferSize;
             OverflowStrategy = overflowStrategy;
-            Buffer = bufferSize != 0 ? Implementation.Buffer.Create<T>(bufferSize, maxFixedBufferSize) : null;
+            Buffer = bufferSize > 0 ? Implementation.Buffer.Create<T>(bufferSize, maxFixedBufferSize) : null;
         }
 
         /// <summary>
         /// TBD
         /// </summary>
-        protected ILoggingAdapter Log => _log ??= Context.GetLogger();
+        protected ILoggingAdapter Log { get; } = Context.GetLogger();
 
         /// <summary>
         /// TBD
@@ -90,7 +90,7 @@ protected bool DefaultReceive(object message)
                 Context.Stop(Self);
             else if (message is Status.Success)
             {
-                if (BufferSize == 0 || Buffer.IsEmpty)
+                if (Buffer is null || Buffer.IsEmpty)
                     OnCompleteThenStop(); // will complete the stream successfully
                 else
                     Context.Become(DrainBufferThenComplete);
@@ -112,7 +112,7 @@ protected virtual bool RequestElement(object message)
             if (message is Request)
             {
                 // totalDemand is tracked by base
-                if (BufferSize != 0)
+                if (Buffer is not null)
                     while (TotalDemand > 0L && !Buffer.IsEmpty)
                         OnNext(Buffer.Dequeue());
 
@@ -133,7 +133,7 @@ protected virtual bool ReceiveElement(T message)
             {
                 if (TotalDemand > 0L)
                     OnNext(message);
-                else if (BufferSize == 0)
+                else if (Buffer is null)
                     Log.Debug("Dropping element because there is no downstream demand: [{0}]", message);
                 else if (!Buffer.IsFull)
                     Buffer.Enqueue(message);
@@ -189,7 +189,7 @@ private bool DrainBufferThenComplete(object message)
                 // even if previously valid completion was requested via Status.Success
                 OnErrorThenStop(failure.Cause);
             }
-            else if (message is Request)
+            else if (message is Request && Buffer is not null)
             {
                 // totalDemand is tracked by base
                 while (TotalDemand > 0L && !Buffer.IsEmpty)
@@ -201,7 +201,7 @@ private bool DrainBufferThenComplete(object message)
             else if (IsActive)
                 Log.Debug(
                     "Dropping element because Status.Success received already, only draining already buffered elements: [{0}] (pending: [{1}])",
-                    message, Buffer.Used);
+                    message, Buffer?.Used ?? 0);
             else
                 return false;
 

src/core/Akka.Streams/Implementation/Buffers.cs

@@ -10,6 +10,7 @@
 using System.Runtime.CompilerServices;
 using Akka.Annotations;
 
+#nullable enable
 namespace Akka.Streams.Implementation
 {
     /// <summary>
@@ -54,7 +55,7 @@ internal interface IBuffer<T>
         /// TBD
         /// </summary>
         /// <returns>TBD</returns>
-        T Peek();
+        T? Peek();
         /// <summary>
         /// TBD
         /// </summary>
@@ -161,7 +162,7 @@ internal abstract class FixedSizeBuffer<T> : IBuffer<T>
         /// </summary>
         protected long WriteIndex;
 
-        private readonly T[] _buffer;
+        private readonly T?[] _buffer;
 
         /// <summary>
         /// TBD
@@ -222,14 +223,20 @@ public void Enqueue(T element)
         /// <param name="element">TBD</param>
         /// <param name="maintenance">TBD</param>
         [MethodImpl(MethodImplOptions.NoInlining)]
-        public void Put(long index, T element, bool maintenance) => _buffer[ToOffset(index, maintenance)] = element;
+        public void Put(long index, T? element, bool maintenance) => _buffer[ToOffset(index, maintenance)] = element;
 
         /// <summary>
         /// TBD
         /// </summary>
         /// <param name="index">TBD</param>
         /// <returns>TBD</returns>
-        public T Get(long index) => _buffer[ToOffset(index, false)];
+        public T Get(long index)
+        {
+            var elem = _buffer[ToOffset(index, false)];
+            if(elem == null)
+                throw new IndexOutOfRangeException($"Invalid buffer element at index {index}, element is null");
+            return elem;
+        }
 
         /// <summary>
         /// TBD
@@ -354,7 +361,7 @@ private sealed class FixedQueue : IBuffer<T>
             private const int Size = 16;
             private const int Mask = 15;
 
-            private readonly T[] _queue = new T[Size];
+            private readonly T?[] _queue = new T[Size];
             private readonly BoundedBuffer<T> _boundedBuffer;
             private int _head;
             private int _tail;
@@ -373,6 +380,9 @@ public FixedQueue(BoundedBuffer<T> boundedBuffer)
 
             public void Enqueue(T element)
             {
+                if(element is null)
+                    throw new ArgumentNullException(nameof(element));
+
                 if (_tail - _head == Size)
                 {
                     var queue = new DynamicQueue(Capacity);
@@ -392,12 +402,15 @@ public T Dequeue()
             {
                 var pos = _head & Mask;
                 var ret = _queue[pos];
+                if(ret is null)
+                    throw new IndexOutOfRangeException();
+
                 _queue[pos] = default(T);
                 _head += 1;
                 return ret;
             }
 
-            public T Peek() => _tail == _head ? default(T) : _queue[_head & Mask];
+            public T? Peek() => _tail == _head ? default(T) : _queue[_head & Mask];
 
             public void Clear()
             {
@@ -431,12 +444,15 @@ public DynamicQueue(int capacity)
 
             public T Dequeue()
             {
+                if(First is null)
+                    throw new IndexOutOfRangeException();
+
                 var result = First.Value;
                 RemoveFirst();
                 return result;
             }
 
-            public T Peek() => First.Value;
+            public T? Peek() => First is null ? default : First.Value;
 
             public void DropHead() => RemoveFirst();
 
@@ -498,7 +514,7 @@ public BoundedBuffer(int capacity)
         /// TBD
         /// </summary>
         /// <returns>TBD</returns>
-        public T Peek() => _q.Peek();
+        public T? Peek() => _q.Peek();
 
         /// <summary>
         /// TBD