SimpleMessageStreamProvider uses a lot of internal classes

[undecided2013]

It cannot be used as the basis for any new custom stream providers in external projects as they won't compile. If it is to be used as the template for new projects, it needs to illustrate usage of the extension points and public apis for creating custom stream providers.

[jason-bragg]

At the moment, we don't really provide much tech to assist stream provider developers. Much of the streaming infrastructure used by the existing stream providers is inaccessible to stream provider developers.

Much of the SimpleMessageStreamProvider (SMS) implementation is internal. I don't believe it was intended to be a template, and has (to my knowledge) no extension points.

The PersistentStreamProvider uses many of the same internals as SMS, and has an adapter extension point, as seen in the AzureQueueStreamProvider, but, like SMS, was not intended to be a template.

Could you offer suggestions of what we could do to improve the stream provider developer story?

[gabikliot]

I would like to second what Jason wrote. Writing a completely new stream provider is very hard now and although in the long run we might want to improve that, it was not our focus so far. Look here for all the options we envisioned for extending streams: http://dotnet.github.io/orleans/Orleans-Streams/Streams-Extensibility

I don't believe it was intended to be a template, and has (to my knowledge) no extension points

The extension points to SMS are via configuration. For Persistent streams it is configuration and Queue Adapters.

[undecided2013]

Thanks, I saw that page and I am starting to dig deeper into the Azure Provider to see what I can do with it. My goal is to add ActiveMQ and Kafka as stream providers.

[randa1]

@undecided2013, regarding Kafka - we at Gigya (specifically @yoniabrahamy) has been working on a Kafka persistent stream provider for quite some time now. It's in the final stages of testing and we're going to expose it with the next stable version of Orleans (currently we use the master branch version). So I'll try to update here when it's public.

[amamh]

@undecided2013

Hey, you could have a look at my custom stream provider here. It's made to use Redis. It's pretty simple and I can help if you have a problem with it.

You can also easily replace Redis with something else for the physical queue. It would be very straight forward to use MQ instead of Redis.

[gabikliot]

@amamh , in your provider code (in the adapter and the factory) you are using a lot of synchronous blocking calls. Synchronous blocking calls are not allowed at all in the stream provider code, since this code runs inside a silo and no synchronous blocking calls are allowed inside the silo.

[amamh]

@gabikliot Thanks for the feedback :)

I'm not sure how I can do that. How would you, for example, establish the connection to Redis in the provider factor in an asynchronous way?

[gabikliot]

Look at AzureQueue stream provider example. We do it there. You can start and check the task later, you can delay until used, different ways. Absolutely no synchronous blocking call is allowed inside the silo, no matter how hard it is to write async code (and it is not hard at all).

[amamh]

@gabikliot

I'm fixing that right now. There is something I don't get in azure provider:

https://github.com/dotnet/orleans/blob/fc0af9079413505de6c5dad928a266fccda74d68/src/OrleansAzureUtils/Providers/Streams/AzureQueue/SimpleAzureQueueAdapter.cs#L86-L94

Why are you checking Queue == null again?

[amamh]

@gabikliot

I've removed all blocking calls in the adapter/receiver. Thanks for the suggestion :)

However, there are still blocking calls in the cache (because I cache on Redis). The factory interface doesn't give me an async cache method:

public IQueueAdapterCache GetQueueAdapterCache()

I wonder why it can't async like the adapter creating method:

public async Task<IQueueAdapter> CreateAdapter()

It would make sense to give the implementer all asynchronous interface methods.

Another inconsistency is in the receiver vs. adapter. The receiver has an async Init method: public Task Initialize(TimeSpan timeout)

but the adapter doesn't have that, so I added one myself which I call in the async CreateAdapter in the factory class.

[gabikliot]

These are all could potentially be good suggestions to extend/change the Adapter and Cache Init/Create APIs. You can open an issue suggesting those and I am sure @jason-bragg will have a lot to say about that.

[jason-bragg]

To reinforce @undecided2013's original point:

Since providers are extension points to the Orleans framework, they should utilize Orleans public interfaces. Any internals of Orleans core or runtime, which we find we need to write providers, should be considered a barrier to others writing providers. That is, if we can't write a provider without use of Orleans internals, how can we expect external developers to do so.

I don't think we can/should perform a mass refactor of the existing streaming infrastructure, but incremental changes over time should get us where we need to go, especially if we can identify the main areas where we rely on internals.

Some areas where we utilize internals:

• Persistent stream pulling agent is a system target.
• Queue Balancer interfaces are internal
• Grain extensions utilize internals

The stream PubSub system interfaces are also internal, but none of its implementation is internal, so users should be able to develop their own similar system. This is not true for the queue balancing infrastructure or the pulling agent.

[jason-bragg]

@gabikliot What is your opinion on changing the Pulling Agent to be a reentrant prefer local placement grain, instead of a system target?

[gabikliot]

It would be hard I think. In case of silo failure they might be re-created on a different silo, for example when pub sub calls into the agent to notify it about new consumers. In general, I think without ensuring real physical grains (see #610 for some related discussion), an abstraction we currently miss in our public API, I think doing this is possible (e.g, via grain self validating its location, etc...) but hard.

Priority wise, I would personally prioritize it lower, then improving the adapter framework, EH adapter, and fixing all the open issues now we have with streaming (there are a couple of open bugs related to streaming).

[sergeybykov]

@randa1 @yoniabrahamy

@undecided2013, regarding Kafka - we at Gigya (specifically @yoniabrahamy) has been working on a Kafka persistent stream provider for quite some time now. It's in the final stages of testing and we're going to expose it with the next stable version of Orleans (currently we use the master branch version). So I'll try to update here when it's public.

Did you end up publishing the Kafka provider?

[sergeybykov]

I believe this has been resolved via adding the EventHub stream provider.