GetAllSubscriptionHandles() breaks implicit subscriptions

[nZeus]

Hello,

We faced an issue related to how StreamConsumerExtension and implicit subscriptions work.
We noticed that invoking stream.GetAllSubscriptionHandles() within our Grain breaks behavior of its implicit subscription(s).

We drilled it down to the fact that GetAllSubscriptionHandles() calls BindExtensionLazy (

orleans/src/Orleans.Streaming/Internal/StreamConsumer.cs

Line 248 in acde8ff

(myExtension, myGrainReference) = providerRuntime.BindExtension<StreamConsumerExtension, IStreamConsumerExtension>(() => new StreamConsumerExtension(providerRuntime));

) which creates StreamConsumerExtension with empty observer (

orleans/src/Orleans.Streaming/Internal/StreamConsumerExtension.cs

Lines 45 to 47 in acde8ff

	internal StreamConsumerExtension(IStreamProviderRuntime providerRt, IStreamSubscriptionObserver streamSubscriptionObserver = null)
	{
	this.streamSubscriptionObserver = streamSubscriptionObserver;

). The observer is not set in any other place.

As a result, messages cannot be delivered, because of this check:

orleans/src/Orleans.Streaming/Internal/StreamConsumerExtension.cs

Line 139 in acde8ff

else if(this.streamSubscriptionObserver != null)

, so we get Dropping on the floor error.

Could you please check why there is such a side-effect of GetAllSubscriptionHandles()?

Kind regards,
Ilia

[nZeus]

I guess it is important to mention that if the IStreamSubscriptionObserver.OnSubscribed was executed before the GetAllSubscriptionHandles() then it works fine, but if we do GetAllSubscriptionHandles() before OnSubscribed is hit, then we get into this issue

[tomachristian]

Another strange thing we found is that there is this

orleans/src/Orleans.Streaming/StreamConsumerGrainContextAction.cs

Line 9 in acde8ff

internal class StreamConsumerGrainContextAction : IConfigureGrainContext

which is registered as

orleans/src/Orleans.Streaming/Hosting/StreamingServiceCollectionExtensions.cs

Line 37 in acde8ff

services.AddSingleton<IConfigureGrainContext, StreamConsumerGrainContextAction>();

which is NOT USED. It might be when things got switched to using

orleans/src/Orleans.Runtime/Activation/IGrainContextActivator.cs

Line 140 in acde8ff

public interface IConfigureGrainContextProvider

like

orleans/src/Orleans.Runtime/Hosting/DefaultSiloServices.cs

Line 113 in acde8ff

services.AddSingleton<IConfigureGrainContextProvider, MayInterleaveConfiguratorProvider>();

, but StreamConsumerGrainContextAction was somehow forgotten.