Do not use IP address 0.0.0.0 on the silo.

[amamh]

Version: 1.0.10 -- latest stable

I 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.

My problem is that I can't subscribe to it from normal non-grain objects.

My example is as follows:

• A producer app: sets some data on the "producer grain"
• A producer grain: creates a stream using my provider and writes to it every time it gets data
• A consumer app: creates a IAsyncObserver object that subscribes to same stream that the producer grain is writing to.
• A zeromq proxy app that acts as the physical queue i.e. my stream is internally using this to communicate.
• Orleans server

It works fine if:

• I replace the consumer app with a consumer grain that does the same thing.
• 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.
• Replace my custom persistent stream with SimpleMessageStreamProvider.

When I debug the source, I can see that it hangs in PersistentStreamPullingAgent.DoHandshakeWithConsumer:

                requestedToken = await AsyncExecutorWithRetries.ExecuteWithRetries(
                     i => consumerData.StreamConsumer.GetSequenceToken(consumerData.SubscriptionId),
                     AsyncExecutorWithRetries.INFINITE_RETRIES,
                     (exception, i) => true,
                     config.MaxEventDeliveryTime,
                     DefaultBackoffProvider);

The source code for this sample project is here:
https://github.com/amamh/orleans-PipeStreamProvider/tree/netmq

[jason-bragg]

To be clear, the code you have now (below) does not work?

from orleans-PipeStreamProvider/Client/Program.cs
public async Task TestObserver.Subscribe()
{
var ccGrain = GrainClient.GrainFactory.GetGrain(0);

    var details = await ccGrain.GetStreamDetails(); 
    var providerName = details.Item1; 
    var nameSpace = details.Item2; 
    var guid = details.Item3; 

    var provider = GrainClient.GetStreamProvider(providerName); 
    var stream = provider.GetStream<int>(guid, nameSpace); 
    await stream.SubscribeAsync(this, null); 
}

[amamh]

@jason-bragg

Yes.

[jason-bragg]

This is the third time I've raised this request. Please add tracing or debug information to your adapter to ensure that data is actually flowing through your adapter.

Unless I'm missing something, zeromq us not a persistent queue. It's a message queue library built over sockets, and you're adapter has hard coded socket addresses. Can you verify that, using your adapter layer only, you can send messages between client and host?

In general, if you believe you've found an issue in the streaming framework, it should be reproducible with the azure or sms stream providers. What would really help is if you could submit a test to our streaming tests that reproduces the issue you are having.

[amamh]

If there is an issue with zmq, how come it works fine when I use an observer grain instead of a normal object?

When debugging, I can see that the messages reach the server fine and it can read them in the receiver, but they never reach the consumer app.

or sms stream providers

I thought the sms stream provider doesn't using the same mechanism as the persistent stream provider? am I mistaken?

---

I'm going to replace zmq with Redis and add logging everywhere as you suggested.

[jason-bragg]

"If there is an issue with zmq, how come it works fine when I use an observer grain instead of a normal object?"

I'm not saying there is anything wrong with zmq. I was questioning the hard coded address. Those would make more sense if you were talking to some persistent queue service. They make less sense in a distributed technology like Orleans where you have no control over which silo is responsible for processing events from a queue.

"I thought the sms stream provider doesn't using the same mechanism as the persistent stream provider? am I mistaken?"

They share a lot of code, especially regarding the pub-sub system and grain extensions. However, you are correct, there are significant differences, like the use of queues and the pulling agent.

"When debugging, I can see that the messages reach the server fine and it can read them in the receiver, but they never reach the consumer app."

Good info.
So you can send data, receive it on the silo, then something goes wrong in the handshake. Because we retry the GetSequenceToken calls (the call that goes to the consumer) we lose any specific exceptions. That's a bit of a bummer.

I think I've enough info to reproduce this. If I can't I'll have more questions.

[amamh]

@jason-bragg

Great. I'm also redoing it using Redis instead, so at least I'm using a "real" persistent queue that I can debug independently, and for fun...

[jason-bragg]

Hmm. I was unable to repro.

Please take a look at: Added tests to subscribe from streams (SMS and AQ) from client. #1023

Does that accurately reproduce your scenario?

[amamh]

@jason-bragg

Ok, I will. Also, I've just replaced zmq with Redis, and I got the exact same result i.e. using an observer grain works but an observer client doesn't work.

[amamh]

@jason-bragg

You once mentioned that I cannot serialise the stream. This was when I was trying to get the stream from the producer grain var stream = await dataGrain.GetStream(). What would I need to do to make it serialisable? I want to try to do that instead of using the provider on the client side.

[jason-bragg]

To be clear, the stream 'is' serializable, but we have no tests around using the serialized stream. It 'should' work, but due to lack of testing, my confidence is not high.

We have done testing around using streams on the client, using the stream acquired from the provider (though not on GitHub, hence my adding them) so I suggested doing it that way. If it had worked it would indicate that there is something wrong with using serialized streams, and then we could look into that. Since it didn't resolve the issue, the cause is still unknown.

Once we've gotten to the root of the client subscription issue you are having, you can try using the serialized stream again, and see if it works. If it does not, we may investigate that, but at this point it is unclear if that is a desirable requirement of the streaming API.

As for the client streaming issue:
Unless you find a deviation in the added tests, I think I'll need silo and client logs to track this further. You've asserted that the adapter is performing as expected. The tests indicate that subscribing to streams using Orleans clients works for persistent stream providers. Those to facts should be sufficient to expect the end to end solution to work.. More information is needed to track what is going wrong.

[amamh]

@jason-bragg

I see.

Wouldn't it be easier to simply run the sample I have on my repo than going through logs? If not, then I'll produce the logs you need and post them.

[jason-bragg]

I don't really have the bandwidth to build, deploy and run/debug other's apps. I'm familiar with the logs Orleans generates, so it's far more efficient for me to just review your logs. In general, you'll need to have sufficient application logging an monitoring for your service anyway, so setting up service logs should not be wasted work on your part.