{"action":"created","issue":{"url":"https://api.github.com/repos/dotnet/orleans/issues/1022","labels_url":"https://api.github.com/repos/dotnet/orleans/issues/1022/labels{/name}","comments_url":"https://api.github.com/repos/dotnet/orleans/issues/1022/comments","events_url":"https://api.github.com/repos/dotnet/orleans/issues/1022/events","html_url":"https://github.com/dotnet/orleans/issues/1022","id":116738554,"number":1022,"title":"Can't get custom PersistentStreamProvider to work with any non-grain observer","user":{"login":"amamh","id":2524201,"avatar_url":"https://avatars.githubusercontent.com/u/2524201?v=3","gravatar_id":"","url":"https://api.github.com/users/amamh","html_url":"https://github.com/amamh","followers_url":"https://api.github.com/users/amamh/followers","following_url":"https://api.github.com/users/amamh/following{/other_user}","gists_url":"https://api.github.com/users/amamh/gists{/gist_id}","starred_url":"https://api.github.com/users/amamh/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/amamh/subscriptions","organizations_url":"https://api.github.com/users/amamh/orgs","repos_url":"https://api.github.com/users/amamh/repos","events_url":"https://api.github.com/users/amamh/events{/privacy}","received_events_url":"https://api.github.com/users/amamh/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":15,"created_at":"2015-11-13T10:09:39Z","updated_at":"2015-11-14T00:52:05Z","closed_at":null,"body":"Version: 1.0.10 -- latest stable\r\n\r\nI made a PersistentStreamProvider with a custom adapter that retains old messages in a cache (inside IQueueCache) and is able to replay it for new clients.\r\n\r\nMy problem is that I can't subscribe to it from normal non-grain objects.\r\n\r\nMy example is as follows:\r\n\r\n- A producer app: sets some data on the \"producer grain\"\r\n- A **producer grain**: creates a stream using my provider and writes to it every time it gets data\r\n- A **consumer app**: creates a **IAsyncObserver object** that subscribes to same stream that the producer grain is writing to.\r\n- A zeromq proxy app that acts as the physical queue i.e. my stream is internally using this to communicate.\r\n- Orleans server\r\n\r\nIt **works fine if**:\r\n- I replace the consumer app with **a consumer grain** that does the same thing.\r\n- Instead of subscribing in the consumer app, I send the consumer object to the producer grain which does the subscription for me, as in `dataGrain.SubscribeMe(this)` where `this` is the consumer object. This means that the **subscription happens on the server side** instead of the client side.\r\n- Replace my custom persistent stream with SimpleMessageStreamProvider.\r\n\r\nWhen I debug the source, I can see that it hangs in PersistentStreamPullingAgent.DoHandshakeWithConsumer:\r\n\r\n\r\n                    requestedToken = await AsyncExecutorWithRetries.ExecuteWithRetries(\r\n                         i => consumerData.StreamConsumer.GetSequenceToken(consumerData.SubscriptionId),\r\n                         AsyncExecutorWithRetries.INFINITE_RETRIES,\r\n                         (exception, i) => true,\r\n                         config.MaxEventDeliveryTime,\r\n                         DefaultBackoffProvider);\r\n\r\n\r\n\r\nThe source code for this sample project is here:\r\nhttps://github.com/amamh/orleans-PipeStreamProvider/tree/netmq"},"comment":{"url":"https://api.github.com/repos/dotnet/orleans/issues/comments/156601080","html_url":"https://github.com/dotnet/orleans/issues/1022#issuecomment-156601080","issue_url":"https://api.github.com/repos/dotnet/orleans/issues/1022","id":156601080,"user":{"login":"amamh","id":2524201,"avatar_url":"https://avatars.githubusercontent.com/u/2524201?v=3","gravatar_id":"","url":"https://api.github.com/users/amamh","html_url":"https://github.com/amamh","followers_url":"https://api.github.com/users/amamh/followers","following_url":"https://api.github.com/users/amamh/following{/other_user}","gists_url":"https://api.github.com/users/amamh/gists{/gist_id}","starred_url":"https://api.github.com/users/amamh/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/amamh/subscriptions","organizations_url":"https://api.github.com/users/amamh/orgs","repos_url":"https://api.github.com/users/amamh/repos","events_url":"https://api.github.com/users/amamh/events{/privacy}","received_events_url":"https://api.github.com/users/amamh/received_events","type":"User","site_admin":false},"created_at":"2015-11-14T00:52:05Z","updated_at":"2015-11-14T00:52:05Z","body":"@jason-bragg \r\n\r\n...... what do you mean by using the same port? they are on the same machine indeed."}}