Support returning IAsyncObservable<T> from Grain methods

[ReubenBond]

This is a feature request. I may implement it, but I am looking for input before I make an attempt.

Currently, Grains can return Task & Task<T> to consumers. This represents the asynchronous nature of distributed systems programming well and encapsulates the possibility of failure. Task<T> is restricted to returning only a single value, though.

Ideally, Grains should be able to return IObservable<T> or IAsyncObservable<T> (depending on whether or not back-pressure/ack is required) in addition to Task & Task<T>.

This differs from the Virtual Streams offered by Orleans in a few ways:

1. Observables are temporary (like tasks), Streams are eternal (like grains)
2. Observables are 1:1 by default, Streams may be many:many
3. Observables can take parameters, Streams are identified by (type, key).

Semantics:

• Observable methods are invoked on SubscribeAsync, not when the client obtains the IAsyncObservable<T>.
• Silo failures are propagated to observers through OnErrorAsync. Clients can at that point re-SubscribeAsync, at which point the grain method is invoked again.

Any obvious issues which aren't covered here?

Input is be greatly appreciated :)

[gabikliot]

What is wrong with Task?
I also totally didn't understand your semantics. I think I don't understand how you propose to use it.

[dVakulen]

Good idea! Some initial thoughts:
Observables may be 1:many, so there will be a need to store multiple subscriber grains references.

Silo failures are propagated to observers through OnErrorAsync

Orleans handles silo failures automatically, also such failures doesn't happen too often, and making user handle it would add seemingly unnecessary boilerplate code.

Observables can take parameters

can you please explain in more detail - what kind of parameters they would take?

[gabikliot]

Commenting from phone!!!
Meant to write what is wrong with Task.

[gabikliot]

Ufff....
Task of IAsyncObs of T as generic parameter.

[ReubenBond]

@gabikliot: Task<IAsyncObservable<T>> is alright, but this feature necessarily requires some core work. IAsyncObservable<T> more clearly represents what's going on and shows that grains can return reactive streams directly.

@dVakulen: I mean to say that Grain interfaces can look like this:

interface ILocationGrain : IGrainWithGuidKey
{
  IAsyncObservable<Deal> FindDeals(SearchFilter filter);
}

I can pass some parameter, filter, to the grain when creating the observable. The grain can hold onto the observer and push updates to it when they arrive as long as they match the filter. So we can implement a standing query.

Another example might be a grain which emits events of different priorities and I only want high-priority events, so I pass that in.

Eventually we might be able to return IAsyncQbservable<T>, but we will look into that when Rx 3.0 is released.

[gabikliot]

So same semantics as Task of IAsyncObs, just different syntax, right? Sure, sounds doable.

[ReubenBond]

Very similar semantics, yes. One difference is that the Grain method which returns the IAsyncObservable<T> is treated similarly to Observable.Create() - it is lazily invoked on subscription. That allows it to be reinvoked upon resubscription (consumer-side retry).

The implementation could involve a consumer-side proxy representing the grain method invocation used to get the IAsyncObservable<T>. A copy of the arguments/etc would be kept in that proxy so that the grain method could be invoked every time SubscribeAsync is called on the consumer-side object.

[yevhen]

LOL. Tell me the truth, are you secretly watching into Orleankka's codebase? 😄

I have this feature implemented already. So that StreamRef can be passed around much the same way like ActorRef.

[gabikliot]

Streams can be passable around in Orleans. IAsyncStream should be passable. If not, it's a bug. As I understood, Reuben asks for something else.

[ReubenBond]

This feature is about extending actor interfaces to support returning streams of values (IAsyncObservable) and would not use the Virtual Streams infrastructure. It's just the logical conclusion of actor interfaces that if I can return a single value asynchronously, I should be able to return multiple values, too.

-----Original Message-----
From: "Gabriel Kliot" notifications@github.com
Sent: ‎10/‎29/‎2015 7:28 AM
To: "dotnet/orleans" orleans@noreply.github.com
Cc: "Reuben Bond" reuben.bond@gmail.com
Subject: Re: [orleans] Support returning IAsyncObservable from Grainmethods (#940)

Streams can be passable around in Orleans. IAsyncStream should be passable. If not, it's a bug. As I understood, Reuben asks for something else.
—
Reply to this email directly or view it on GitHub.

[yevhen]

Understood. Like temporary channels? Isn't is what observers were for?

[jthelin]

+1 Nice idea @ReubenBond

This feature is about extending actor interfaces to support returning streams of values (IAsyncObservable)

I am currently working on another project that would greatly benefit from having this feature available in Orleans too.

[gabikliot]

Just to be clear again: we can, now, without any changes, return from the grain method Task<IAsyncStream<T>>, and since IAsyncStream<T> is also IAsyncObservable<T>, we can return Task<IAsyncObservable<T>>. So we already support "returning streams of values".

@ReubenBond is asking for more. As I understood, he is asking for syntactic sugar of returning IAsyncObservable<T> instead of Task<IAsyncStream<T>> and being able to subscribe to the returned IAsyncObservable immediately, without first awaiting the Task.

[ReubenBond]

@gabikliot just to be clear: returning Task<IAsyncStream<T>, etc, only work if that IAsyncStream<T> is a virtual stream reference - right?

[gabikliot]

Well, currently, we have only one implementation of the IAsyncStream<T> interface, so yes.

We do actually need to create a test and check it. It "should" work, as it was designed this way, but I don't remember having a test for that, so it is possible there is some field we don't serialize properly or don't reconstruct properly when deserializing.

[ReubenBond]

I want to be able to return physical streams: streams which will break when the grain is deactivated, forcing the user to resubscribe in order to continue receiving values.

For example, I might want to return results to a search query from the grain. Eg, imagine I have an IChatRoomGrain and I want to return only messages which satisfy a custom query, for example, messages which contain the user's name. The client can include the last known messageId in the request to ensure they don't miss anything.

More generally when Bonsai is released, I want methods which take an expression, apply it to a stream (eg, traffic reports), and return the resulting stream.

Does that sound like desirable functionality to you?

[gabikliot]

Yes, it does. I thought quite a lot about a similar scenario, recently and also originally, when we just started thinking about streams. I will open a new issue and describe my thoughts on that, but a bit later (give me a day or two).

[danvanderboom]

@ReubenBond It sounds like this idea is to Timers what Virtual Streams are to Reminders. We need both durable and non-durable versions of each of these core concepts. A grain instance stream sounds very useful, in addition to the grain streams currently supported.