Fix fail test case in QueueSourceSpec and added StreamDetachedExcepti…

…on when stream already completed

src/core/Akka.Streams.Tests/Dsl/QueueSinkSpec.cs

@@ -153,7 +153,7 @@ public void QueueSink_should_timeout_future_when_stream_cannot_provide_data()
             }, _materializer);
         }
 
-        [Fact(Skip = "Racy, see https://github.com/akkadotnet/akka.net/pull/4424#issuecomment-632284459")]
+        [Fact]
         public void QueueSink_should_fail_pull_future_when_stream_is_completed()
         {
             this.AssertAllStagesStopped(() =>
@@ -170,10 +170,8 @@ public void QueueSink_should_fail_pull_future_when_stream_is_completed()
                 var result = queue.PullAsync().Result;
                 result.Should().Be(Option<int>.None);
 
-                ((Task)queue.PullAsync()).ContinueWith(t =>
-                {
-                    t.Exception.InnerException.Should().BeOfType<IllegalStateException>();
-                }, TaskContinuationOptions.OnlyOnFaulted).Wait();
+                var exception = Record.ExceptionAsync(async () => await queue.PullAsync()).Result;
+                exception.Should().BeOfType<StreamDetachedException>();
             }, _materializer);
         }
 

src/core/Akka.Streams.Tests/Dsl/QueueSourceSpec.cs

@@ -338,11 +338,12 @@ public void QueueSource_should_fail_offer_future_when_stream_is_completed()
                         .Run(_materializer);
                 var sub = s.ExpectSubscription();
 
-                queue.WatchCompletionAsync().ContinueWith(t => "done").PipeTo(TestActor);
+                queue.WatchCompletionAsync().ContinueWith(t => Done.Instance).PipeTo(TestActor);
                 sub.Cancel();
-                ExpectMsg("done");
+                ExpectMsg(Done.Instance);
 
-                queue.OfferAsync(1).ContinueWith(t => t.Exception.Should().BeOfType<IllegalStateException>());
+                var exception = Record.ExceptionAsync(async () => await queue.OfferAsync(1)).Result;
+                exception.Should().BeOfType<StreamDetachedException>();
             }, _materializer);
         }
 

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

@@ -624,10 +624,10 @@ public override void OnUpstreamFailure(Exception e)
 
             public override void PostStop()
             {
-                if(!_completionSignalled)
+                if (!_completionSignalled)
                     _promise.TrySetException(new AbruptStageTerminationException(this));
             }
-            
+
             public override void PreStart() => Pull(_stage.In);
         }
 
@@ -805,12 +805,8 @@ public override void PreStart()
                 Pull(_stage.In);
             }
 
-            public override void PostStop()
-            {
-                StopCallback(
-                    promise =>
-                            promise.SetException(new IllegalStateException("Stream is terminated. QueueSink is detached")));
-            }
+            public override void PostStop() => 
+                StopCallback(promise => promise.SetException(StreamDetachedException.Instance));
 
             private Action<TaskCompletionSource<Option<T>>> Callback()
             {
@@ -1024,7 +1020,8 @@ private void InitInternalSource(Sink<TIn, TMat> sink, TIn firstElement)
             {
                 var sourceOut = new SubSource(this, firstElement);
 
-                try {
+                try
+                {
                     var matVal = Source.FromGraph(sourceOut.Source)
                         .RunWith(sink, Interpreter.SubFusingMaterializer);
                     _completion.TrySetResult(matVal);
@@ -1034,7 +1031,7 @@ private void InitInternalSource(Sink<TIn, TMat> sink, TIn firstElement)
                     _completion.TrySetException(ex);
                     FailStage(ex);
                 }
-                
+
             }
 
             #region SubSource
@@ -1176,7 +1173,7 @@ public void Dispose(bool unregister)
                 }
             }
         }
-        
+
         private sealed class ObservableLogic : GraphStageLogic, IObservable<T>
         {
             private readonly ObservableSinkStage<T> _stage;
@@ -1216,12 +1213,12 @@ public void Remove(IObserver<T> observer)
                 ImmutableInterlocked.TryRemove(ref _observers, observer, out var _);
             }
 
-            public IDisposable Subscribe(IObserver<T> observer) => 
+            public IDisposable Subscribe(IObserver<T> observer) =>
                 ImmutableInterlocked.GetOrAdd(ref _observers, observer, new ObserverDisposable(this, observer));
         }
 
         #endregion
-        
+
 
         public ObservableSinkStage()
         {

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

@@ -168,7 +168,7 @@ public override void PreStart()
 
             public override void PostStop()
             {
-                var exception = new AbruptStageTerminationException(this);
+                var exception = StreamDetachedException.Instance;
                 _completion.TrySetException(exception);
                 StopCallback(input =>
                 {

src/core/Akka.Streams/StreamTcpException.cs

@@ -44,6 +44,22 @@ protected StreamTcpException(SerializationInfo info, StreamingContext context) :
 #endif
     }
 
+    /// <summary>
+    /// This exception signals that materialized value is already detached from stream. This usually happens
+    /// when stream is completed and an ActorSystem is shut down while materialized object is still available.
+    /// </summary>
+    public class StreamDetachedException : Exception
+    {
+        /// <summary>
+        /// Initializes a single instance of the <see cref="StreamDetachedException"/> class.
+        /// </summary>
+        public static readonly StreamDetachedException Instance = new StreamDetachedException();
+
+        private StreamDetachedException() : base("Stream is terminated. Materialized value is detached.")
+        {
+        }
+    }
+
     /// <summary>
     /// TBD
     /// </summary>