Interop between Akka.Streams and IObservable #3112

Horusiath · 2017-09-16T20:40:20Z

Motivation

Currently Akka.Streams is pretty self-contained - Rx.NET Observables adoption is much wider: partially probably due to fact that IObservable<T> is part of standard library and reactive streams are much younger and not so well-known. As it doesn't bring any extra dependencies, and may help with interop with existing technologies (i.e. SignalR for .NET Core 2.0 supports observables natively), I've decided to make this PR.

API

This PR aims to introduce 2 new methods:

Sink.AsObservable<T>() (already implemented with tests) - allows to materialize stream as IObservable<T>, which can be used for multi-observers scenarios (internally it's using immutable structures + interlocked, so it should be thread-safe as well, but that's harder to check).
Source.FromObservable<T>(IObservable<T> observable, int maxCapacity, OverflowStrategy strategy) (in progress) it works in a similar way to how Source.FromEvent works right now - actually I want to merge those implementations together as C# events can be described as observables easily.

alexvaluyskiy · 2017-09-17T04:49:59Z

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

@@ -1003,7 +1003,7 @@ private Action<TEventArgs> SetupOverflowStrategy(OverflowStrategy overflowStrate
        /// <param name="addHandler">Action which attaches given event handler to the underlying .NET event.</param>
        /// <param name="removeHandler">Action which detaches given event handler to the underlying .NET event.</param>
        /// <param name="maxBuffer">Maximum size of the buffer, used in situation when amount of emitted events is higher than current processing capabilities of the downstream.</param>
-        /// <param name="overflowStrategy">Overflow strategy used, when buffer (size specified by <paramref name="maxBuffer"/>) has been overflown.</param>
+        /// <param name="overflowStrategy">Overflow strategy used, when b{uffer (size specified by <paramref name="maxBuffer"/>) has been overflown.</param>


Are you sure?

alexvaluyskiy · 2017-09-17T04:52:29Z

Related #2354

alexvaluyskiy · 2017-09-17T04:54:31Z

@Horusiath Probably we need an integration with System.Threading.Tasks.Channels in the future too

Horusiath · 2017-09-17T15:12:32Z

When building Source.FromObservable I've run into following problem - since observables don't have any notion of demand/backpressure, they got pretty straightforward way of working order of OnNext/OnError/OnCompleted signals:

var subject = new Subject<int>();
subject.Subscribe(
    onNext: x => Console.WriteLine($"Event: {x}"), 
    onError: e => Console.WriteLine($"Error: {e.Message}"), 
    onCompleted: () => Console.WriteLine("Completed"));

subject.OnNext(1);
subject.OnError(new Exception("hello"));
subject.OnNext(2);

// this will produce: 
//  - Event: 1
//  - Error: hello

When using reactive streams, we do have a backpressure - this means that no new events will be send downstream until an explicit demand will be signalized using subscription.Request(n). However backpressure works only on events! This means that errors and completion signals are being send using a side lane, as soon as they come. This means that errors and completions may came out of order.

What does it mean in the current implementation?:

var subject = new Subject<int>();
var probe = this.CreateManualSubscriberProbe<int>();
Source.FromObservable(subject).To(Sink.FromSubscriber(probe)).Run(_materializer);

var subscription = probe.ExpectSubscription();

subject.OnNext(1);
subject.OnError(new Exception("hello"));
subject.OnNext(2);

subscription.Request(2);

// what user may expect:
subscription.ExpectNext(1); // this won't happen as error is not backpressured and will be delivered immediatelly
subscription.ExpectError();

Ofc, there's an alternative approach - we could buffer errors/completions just like we buffer normal events - but this would complicate buffering and overflow strategy.

I think, that we should allow current behavior, simply because subscription.Request should happen immediately after subscribing to an observable.

alexvaluyskiy · 2017-09-17T17:05:07Z

I think a documentation for this integration should be a part of the PR

Aaronontheweb · 2017-09-19T14:25:35Z

@Horusiath is this still in a "do not merge" state or is it ready for review?

Horusiath · 2017-09-19T14:39:46Z

@Aaronontheweb it's ready to be reviewed. There are some issues with Mono runtime (as usual).

Aaronontheweb · 2017-09-19T15:24:44Z

There are some issues with Mono runtime (as usual).

Grrr... I'll take a look at this and see what I can dig up on those Mono issues.

marcpiechura · 2017-09-20T21:03:41Z

docs/articles/streams/integration.md

+    .RunForEach(Console.WriteLine, materializer);
+```
+
+You may notice two extra parameters here. One of the advantages of Akka.Streams (and reactive streams in general) over Reactive Extensions in notion of backpressure - absent in Rx.NET. This puts a constraint of rate limiting the events incoming form upstream. If an observable will be producing events faster, than downstream is able to consume them, source stage will start to buffer them up to a provided `maxBufferCapacity` limit. Once that limit is reached, an overflow strategy will be applied. There are several different overflow strategies to choose from:


Is “in notion” correct? Sounds a bit strange to me or should it have been “is notion” ?

marcpiechura · 2017-09-20T21:18:03Z

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

+                        foreach (var disposer in old.Values) disposer.Dispose(unregister: false);
+                    },
+                    onUpstreamFailure: e =>
+                    {


Don’t we need to remove observers in case of failure too?

Are we 100% sure that upstream failure should cause removal? What about Observable.Catch<>?

Even if the observable catches the failure the Sink is dead so it wouldn’t produce any new items

marcpiechura · 2017-09-20T21:24:11Z

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

-            _maxBuffer = maxBuffer;
-            _overflowStrategy = overflowStrategy;
-            Shape = new SourceShape<TEventArgs>(Out);
+            this._observable = observable;


Maybe want to remove “this”, haven’t used it anywhere else so...

Horusiath · 2017-09-28T05:18:49Z

I couldn't debug this shit to identify the Mono issue. Any ideas?

alexvaluyskiy · 2017-09-28T07:06:48Z

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

+                return new TestDisposer(this);
+            }
+
+            public void Event(T element) => Observer.OnNext(element);


Try to put `[MethodImpl(MethodImplOptions.NoInlining)] here. It could on Mono

nesc58 · 2018-02-13T16:36:18Z

Something new? This interop would be very nice.

Aaronontheweb · 2018-02-13T18:16:45Z

@neekgreen we need to disable Mono compilation on these builds. I'm sure the Mono runtime can run with these changes just fine, but the Mono compiler cannot.

vasily-kirichenko · 2018-09-12T13:02:33Z

Any plans to include it to 1.4?

Horusiath · 2018-09-12T14:47:27Z

I'll rebase it and push again - the only reason why it didn't land so far was some unrecognized issue on linux/mono.

Aaronontheweb · 2018-10-23T20:26:45Z

Since this was merged into dev, it's going into the v1.3.10 release 🥂

Aaronontheweb added the needs review label Sep 16, 2017

alexvaluyskiy reviewed Sep 17, 2017

View reviewed changes

alexvaluyskiy added the akka-streams label Sep 17, 2017

Horusiath force-pushed the observable-streams branch from 2ed4e29 to f6a93ce Compare September 17, 2017 15:33

Horusiath changed the title ~~[WIP] Interop betweek Akka.Streams and IObservable~~ Interop betweek Akka.Streams and IObservable Sep 17, 2017

Horusiath added the ready label Sep 17, 2017

marcpiechura approved these changes Sep 20, 2017

View reviewed changes

marcpiechura added reviewed positively and removed needs review labels Sep 20, 2017

Horusiath force-pushed the observable-streams branch 2 times, most recently from d2b84c9 to fa93f9a Compare September 27, 2017 21:39

alexvaluyskiy reviewed Sep 28, 2017

View reviewed changes

Horusiath force-pushed the observable-streams branch from 794a415 to c91d08e Compare October 21, 2017 12:17

Horusiath force-pushed the observable-streams branch from 78007a9 to 7434221 Compare January 7, 2018 17:25

Danthar added this to the 1.4.0 milestone Apr 18, 2018

Horusiath added 3 commits October 4, 2018 19:43

implemented Sink.AsObservable

ad11187

initial work on Source.FromObservable

6b14217

tweaks in observable source specs

c75fc29

Horusiath added 5 commits October 4, 2018 19:43

replaced EventSourceStage with ObservableSourceStage

f0c77ef

API approvals and docs

55078e6

more detailed logging

58d3d58

minor improvements

0fa8418

added AssertAllStagesStopped

e3ec658

Horusiath force-pushed the observable-streams branch from 7434221 to e3ec658 Compare October 4, 2018 18:06

Merge branch 'dev' into observable-streams

0e5b8ac

Danthar merged commit 771103c into akkadotnet:dev Oct 22, 2018

Aaronontheweb modified the milestones: 1.4.0, 1.3.10 Oct 23, 2018

Aaronontheweb changed the title ~~Interop betweek Akka.Streams and IObservable~~ Interop between Akka.Streams and IObservable Nov 1, 2018

This was referenced Nov 1, 2018

added v1.3.10 release notes #3638

Merged

v1.3.10 Production Release #3643

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Interop between Akka.Streams and IObservable #3112

Interop between Akka.Streams and IObservable #3112

Horusiath commented Sep 16, 2017 •

edited

Loading

alexvaluyskiy Sep 17, 2017

alexvaluyskiy commented Sep 17, 2017

alexvaluyskiy commented Sep 17, 2017

Horusiath commented Sep 17, 2017

alexvaluyskiy commented Sep 17, 2017

Aaronontheweb commented Sep 19, 2017

Horusiath commented Sep 19, 2017

Aaronontheweb commented Sep 19, 2017

marcpiechura Sep 20, 2017

marcpiechura Sep 20, 2017

Horusiath Sep 21, 2017 •

edited

Loading

marcpiechura Sep 21, 2017

marcpiechura Sep 20, 2017

Horusiath commented Sep 28, 2017

alexvaluyskiy Sep 28, 2017

nesc58 commented Feb 13, 2018

Aaronontheweb commented Feb 13, 2018

vasily-kirichenko commented Sep 12, 2018

Horusiath commented Sep 12, 2018

Aaronontheweb commented Oct 23, 2018

Interop between Akka.Streams and IObservable #3112

Interop between Akka.Streams and IObservable #3112

Conversation

Horusiath commented Sep 16, 2017 • edited Loading

Motivation

API

alexvaluyskiy Sep 17, 2017

Choose a reason for hiding this comment

alexvaluyskiy commented Sep 17, 2017

alexvaluyskiy commented Sep 17, 2017

Horusiath commented Sep 17, 2017

alexvaluyskiy commented Sep 17, 2017

Aaronontheweb commented Sep 19, 2017

Horusiath commented Sep 19, 2017

Aaronontheweb commented Sep 19, 2017

marcpiechura Sep 20, 2017

Choose a reason for hiding this comment

marcpiechura Sep 20, 2017

Choose a reason for hiding this comment

Horusiath Sep 21, 2017 • edited Loading

Choose a reason for hiding this comment

marcpiechura Sep 21, 2017

Choose a reason for hiding this comment

marcpiechura Sep 20, 2017

Choose a reason for hiding this comment

Horusiath commented Sep 28, 2017

alexvaluyskiy Sep 28, 2017

Choose a reason for hiding this comment

nesc58 commented Feb 13, 2018

Aaronontheweb commented Feb 13, 2018

vasily-kirichenko commented Sep 12, 2018

Horusiath commented Sep 12, 2018

Aaronontheweb commented Oct 23, 2018

Horusiath commented Sep 16, 2017 •

edited

Loading

Horusiath Sep 21, 2017 •

edited

Loading