{"action":"opened","issue":{"url":"https://api.github.com/repos/dotnet/orleans/issues/2741","repository_url":"https://api.github.com/repos/dotnet/orleans","labels_url":"https://api.github.com/repos/dotnet/orleans/issues/2741/labels{/name}","comments_url":"https://api.github.com/repos/dotnet/orleans/issues/2741/comments","events_url":"https://api.github.com/repos/dotnet/orleans/issues/2741/events","html_url":"https://github.com/dotnet/orleans/issues/2741","id":208599305,"number":2741,"title":"Programmable stream subscribe API","user":{"login":"xiazen","id":20803801,"avatar_url":"https://avatars.githubusercontent.com/u/20803801?v=3","gravatar_id":"","url":"https://api.github.com/users/xiazen","html_url":"https://github.com/xiazen","followers_url":"https://api.github.com/users/xiazen/followers","following_url":"https://api.github.com/users/xiazen/following{/other_user}","gists_url":"https://api.github.com/users/xiazen/gists{/gist_id}","starred_url":"https://api.github.com/users/xiazen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/xiazen/subscriptions","organizations_url":"https://api.github.com/users/xiazen/orgs","repos_url":"https://api.github.com/users/xiazen/repos","events_url":"https://api.github.com/users/xiazen/events{/privacy}","received_events_url":"https://api.github.com/users/xiazen/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2017-02-18T01:26:15Z","updated_at":"2017-02-18T01:26:15Z","closed_at":null,"body":"## Problem Space\r\nCurrently in Orleans streaming space, there’s two subscribe models supported. One is ImplicitSubscribe, which define the subscription map for orleans streaming infrastructure in compile time. Another one is ExplicitSubscribe, which rely on the consumer grain to be activated to Subscribe and attach its observer to streaming infrastructure. These two model serves great so far and covers most of the use cases in Orleans streaming. But we recently got a new use case, which is the drive for this new feature in Orleans streaming. \r\nThe new use case is:\r\n1.\tI want to dynamically configure consumer’s subscribe/unsubscribe in runtime\r\n2.\tI don’t want to use ExplicitSubscribe model, which requires application to activate the consumer grains to build the subscription map. I want to be able to define the subscription map outside grain context(without async grain call), for efficiency concern. For example, when application boostraping. \r\n3.\tSince my consumers and producers exists in the same silo, I want they to use local streaming infra, for efficiency concern. \r\n\r\nFor this new use case, either of orleans streaming system’s current subscribe models can support it. This drove us to design this new feature for streaming. As follows it the design for this feature. \r\nNew Component \r\n•\t`IStreamSubscriptionManager`: the public interface which application use to configure subscription programmablly.  User can access `StreamSubscriptionManager ` through stream provider. It will not support adding or removing implicit subscriptions.  Those subscriptions will be query only.\r\n\r\n## New Pattern\r\n1.\tApplication configure subscription map using `IStreamSubscriptionManager \r\n`2.\tConsumer grain attach its observer using `ResumeAsync` in its grain logic. \r\n\r\n## Common Questions\r\n1 What would a quick remove/re-add pair do to an active grain? Would it just stop receiving events and then start receiving them again?\r\nIn the current design, it would simply stop delivering events to the consumer, then resume delivering events.  The effects of this ‘interruption’ would depend on the stream provider implementation.  For SMS, any messages sent between the remove/add, could be lost.  For recoverable stream providers like EventHub, when the re-added subscription is being established the handshake with the grain will inform the streaming system where to start delivering from, and if the data is still in the cache no data loss should occur. \r\n\r\n## API Design\r\n\r\nIStreamSubscriptionManager\r\n\r\n```\r\npublic interface IStreamSubscriptionManager\r\n    {\r\n        Task<StreamSubscription> AddSubscription(IStreamIdentity streamId, GrainReference grainRef);\r\n        Task RemoveSubscription(Guid subscriptionId);\r\n        Task<IEnumerable<StreamSubscription>> GetSubscriptions(IStreamIdentity StreamId);\r\n    }\r\n```\r\n\r\nStreamSubscription\r\n```\r\n    public class StreamSubscription\r\n    {\r\n        StreamSubscription(Guid subscriptionId, string streamProviderName, IStreamIdentity streamId, IGrainIdentity grainId)\r\n        {\r\n            this.SubscriptionId = subscriptionId;\r\n            this.StreamProviderName = streamProviderName;\r\n            this.StreamId = streamId;\r\n            this.GrainId = grainId;\r\n        }\r\n        public Guid SubscriptionId;\r\n        public string StreamProviderName;\r\n        public IStreamIdentity StreamId;\r\n        public IGrainIdentity GrainId;\r\n    }\r\n```\r\n\r\n\r\nIStreamProvider\r\nA property will be added to a stream provider for users to access it’s subscription manager.\r\n\r\n```\r\n    public interface IStreamProvider\r\n    {\r\n        string Name { get; }\r\n        IAsyncStream<T> GetStream<T>(Guid streamId, string streamNamespace);\r\n        bool IsRewindable { get; }\r\n        IStreamSubscriptionManager SubscriptionManager { get; }\r\n    }\r\n```\r\n\r\nStreamSubscriptionManagerExtensions\r\n\r\n```\r\npublic static class StreamSubscriptionManagerExtensions\r\n    {\r\n        public static Task<StreamSubscription> AddSubscription<TGrainInterface>(\r\n            this IStreamSubscriptionManager manager,\r\n            IStreamIdentity streamId,\r\n            Guid primaryKey,\r\n            string grainClassNamePrefix = null)\r\n            where TGrainInterface : IGrainWithGuidKey\r\n        {\r\n            // placeholder\r\n            return Task.FromResult<StreamSubscription>(null);\r\n        }\r\n\r\n        public static Task<StreamSubscription> AddSubscription<TGrainInterface>(\r\n            this IStreamSubscriptionManager manager,\r\n            IStreamIdentity streamId,\r\n            long primaryKey,\r\n            string grainClassNamePrefix = null)\r\n            where TGrainInterface : IGrainWithIntegerKey\r\n        {\r\n            // placeholder\r\n            return Task.FromResult<StreamSubscription>(null);\r\n        }\r\n\r\n        public static Task<StreamSubscription> AddSubscription<TGrainInterface>(\r\n            this IStreamSubscriptionManager manager,\r\n            IStreamIdentity streamId,\r\n            string primaryKey,\r\n            string grainClassNamePrefix = null)\r\n            where TGrainInterface : IGrainWithStringKey\r\n        {\r\n            // placeholder\r\n            return Task.FromResult<StreamSubscription>(null);\r\n        }\r\n\r\n        public static Task<StreamSubscription> AddSubscription<TGrainInterface>(\r\n            this IStreamSubscriptionManager manager,\r\n            IStreamIdentity streamId,\r\n            Guid primaryKey,\r\n            string keyExtension,\r\n            string grainClassNamePrefix = null)\r\n            where TGrainInterface : IGrainWithGuidCompoundKey\r\n        {\r\n            // placeholder\r\n            return Task.FromResult<StreamSubscription>(null);\r\n        }\r\n\r\n        public static Task<StreamSubscription> AddSubscription<TGrainInterface>(\r\n            this IStreamSubscriptionManager manager,\r\n            IStreamIdentity streamId,\r\n            long primaryKey,\r\n            string keyExtension,\r\n            string grainClassNamePrefix = null)\r\n            where TGrainInterface : IGrainWithIntegerCompoundKey\r\n        {\r\n            // placeholder\r\n            return Task.FromResult<StreamSubscription>(null);\r\n        }\r\n    }\r\n```\r\n\r\n\r\n## Implementation details\r\n\r\nStreamSubscriptionManager\r\nAn internal StreamSubscriptionManager will be developed that is constructed with the IStreamPubSub configured for the stream provider.\r\n\r\nMemoryPubSub\r\nWe’ll add an implementation of IStreamPubSub that keeps subscription information in local memory.\r\n"}}
{"actor":{"display_login":"xiazen"}}