{"action":"created","issue":{"url":"https://api.github.com/repos/dotnet/orleans/issues/1916","repository_url":"https://api.github.com/repos/dotnet/orleans","labels_url":"https://api.github.com/repos/dotnet/orleans/issues/1916/labels{/name}","comments_url":"https://api.github.com/repos/dotnet/orleans/issues/1916/comments","events_url":"https://api.github.com/repos/dotnet/orleans/issues/1916/events","html_url":"https://github.com/dotnet/orleans/pull/1916","id":164466549,"number":1916,"title":"WIP: Ad-hoc observables for grains (#940)","user":{"login":"ReubenBond","id":203839,"avatar_url":"https://avatars.githubusercontent.com/u/203839?v=3","gravatar_id":"","url":"https://api.github.com/users/ReubenBond","html_url":"https://github.com/ReubenBond","followers_url":"https://api.github.com/users/ReubenBond/followers","following_url":"https://api.github.com/users/ReubenBond/following{/other_user}","gists_url":"https://api.github.com/users/ReubenBond/gists{/gist_id}","starred_url":"https://api.github.com/users/ReubenBond/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ReubenBond/subscriptions","organizations_url":"https://api.github.com/users/ReubenBond/orgs","repos_url":"https://api.github.com/users/ReubenBond/repos","events_url":"https://api.github.com/users/ReubenBond/events{/privacy}","received_events_url":"https://api.github.com/users/ReubenBond/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/dotnet/orleans/labels/cla-already-signed","name":"cla-already-signed","color":"009800"},{"url":"https://api.github.com/repos/dotnet/orleans/labels/enhancement","name":"enhancement","color":"84b6eb"},{"url":"https://api.github.com/repos/dotnet/orleans/labels/work-in-progress","name":"work-in-progress","color":"fad8c7"}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":20,"created_at":"2016-07-08T06:21:39Z","updated_at":"2016-07-13T11:46:36Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/dotnet/orleans/pulls/1916","html_url":"https://github.com/dotnet/orleans/pull/1916","diff_url":"https://github.com/dotnet/orleans/pull/1916.diff","patch_url":"https://github.com/dotnet/orleans/pull/1916.patch"},"body":"*Very* early work-in-progress of ad-hoc observables for grains, as described in #940.\r\n\r\nThis allows users to define grain interfaces like so:\r\n```C#\r\npublic interface IReactiveGrain<T> : IGrainWithGuidKey\r\n{\r\n    IGrainObservable<T> GetStream(T[] values);\r\n}\r\n```\r\n```C#\r\npublic class ReactiveGrain<T> : Grain, IReactiveGrain<T>\r\n{\r\n    public IGrainObservable<T> GetStream(T[] values)\r\n    {\r\n        return Observable.Create<T>(async observer =>\r\n        {\r\n            if (values == null) throw new ArgumentNullException(nameof(values));\r\n            try\r\n            {\r\n                foreach (var value in values) await observer.OnNext(value);\r\n                await observer.OnCompleted();\r\n            }\r\n            catch (Exception exception)\r\n            {\r\n                await observer.OnError(exception);\r\n            }\r\n\r\n            return new AsyncDisposable();\r\n        });\r\n    }\r\n}\r\n```\r\n\r\nMore advanced usage could use filter values or expression trees as parameters.\r\n\r\nThere are obvious deficiencies, some of which are called out in the code. For example, observables should error or be disposed when the remote side dies (depending on which side dies). ~~Additionally, this only works for `int` at the moment. It's an easy fix with a bit of IL.~~ Please raise any concerns you have.\r\n\r\nThis is definitely not a replacement for Virtual Streams, this is a complimentary feature.\r\n\r\nThe remote grain is only called when the returned observable is subscribed to, it is called every time `Subscribe` is called. The observable captures the arguments at the time of creation so that it can subscribe and resubscribe as needed.\r\n\r\nLet me know what you think - all criticism is welcome."},"comment":{"url":"https://api.github.com/repos/dotnet/orleans/issues/comments/232331824","html_url":"https://github.com/dotnet/orleans/pull/1916#issuecomment-232331824","issue_url":"https://api.github.com/repos/dotnet/orleans/issues/1916","id":232331824,"user":{"login":"ReubenBond","id":203839,"avatar_url":"https://avatars.githubusercontent.com/u/203839?v=3","gravatar_id":"","url":"https://api.github.com/users/ReubenBond","html_url":"https://github.com/ReubenBond","followers_url":"https://api.github.com/users/ReubenBond/followers","following_url":"https://api.github.com/users/ReubenBond/following{/other_user}","gists_url":"https://api.github.com/users/ReubenBond/gists{/gist_id}","starred_url":"https://api.github.com/users/ReubenBond/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ReubenBond/subscriptions","organizations_url":"https://api.github.com/users/ReubenBond/orgs","repos_url":"https://api.github.com/users/ReubenBond/repos","events_url":"https://api.github.com/users/ReubenBond/events{/privacy}","received_events_url":"https://api.github.com/users/ReubenBond/received_events","type":"User","site_admin":false},"created_at":"2016-07-13T11:46:36Z","updated_at":"2016-07-13T11:46:36Z","body":"Latest revision eschews `IGrainObserver`, etc, in favor of the existing `IAsyncObservable`/ `IAsyncObserver`/`StreamSubscriptionHandle` types in order to align with Virtual Streams. It adds support for resuming streams after deactivation.\r\n\r\nThe `IAsyncObserver<T>`, `IAsyncObserver<T>` and `StreamSubscriptionHandle<T>` implementations are serializable, so each of them can be stored and reused later. Those entities cannot be transferred between grains/clients yet. So you cannot yet pass observers/observables/subscriptions between grains as arguments. I don't plan to implement that in this PR.\r\n\r\nIf a grain subscribes to another grain, it will be reactivated when it receives a new message. Similarly to Virtual Streams, grains need to call `ResumeAsync(...)` to resume the subscription. If the producer was deactivated, it will be reactivated at that point.\r\n\r\nCalling `ResumeAsync(...)` on the `StreamSubscriptionHandle<T>` locally results in a subscribe on the remote grain's `IObservableGrainExtension`, passing in the existing `streamId`. If the remote grain does not hold a reference to the `StreamSubscriptionHandle` matching that `streamId`, then the grain's method will be called and a new subscription will be started with the existing `streamId`. In order to implement this, the `StreamSubscriptionHandle<T>` implementation holds a reference to the `IAsyncObservable<T>` impl, which holds a reference to the `InvokeMethodRequest` for the original grain method call.\r\n\r\n@jdom wrote:\r\n> But as a benefit is that you could ALSO have each value be processed as a new grain request without reentrancy, which is not something I can see in the IGrainObservable example, unless I'm missing something. In your examples I do not see how you are dealing with reentrancy or asynchronous processing, I'm only seeing that you buffer all the values and when you think you are done receiving then you process all of them in one go (in the case of the test client being the observer) or you mutate the list of chat messages potentially violating reentrancy rules for that grain.\r\n\r\nThis system works via grain extensions, which are processed in the context of a grain call. Orleans doesn't control client, but I *think* calls to client objects are invoked in a single-threaded context. Does anybody know for sure?\r\n\r\nThe feedback is very useful, keep it coming :smile: "}}
{"actor":{"display_login":"ReubenBond"}}