Result of subscribing to topic inaccessible when calling makeSubscription causing race condition

[jrogers]

Currently when you subscribe to a topic via makeSubscription, there's no way to know if/when the subscription succeeded, as the observable returned only provides the PubSubData when something is published to the topic.

I have the following scenario where I create a "dynamic" topic, Ie. the topic has some filter information associated with it. So basically I subscribe to the topic from my client with a generated topic name, then I call a procedure registered on the router supplying the topic URI and my filter information, allowing the publisher to start publishing data to my dynamic topic. This results in a race though. Consider this:

1) Client calls WampClient.makeSubscription with "topic_uri", WampClient sends the subscription request on the event loop thread. Subscription result processed on WampClient scheduler thread. 2) Client calls "createSubscription" procedure via WampClient.call that has been registered, providing "topic_uri" and some filter information, request sent on the event loop thread. 3) Other client that provides "createSubscription" immediately publishes information to "topic_uri", pubsubdata received on event loop thread.

Now the published event is received by the subscribed client, but can dropped because subscription on client has not been fully created, due to the result of the initial subscription request being processed on another thread in WampClient (scheduler) where it adds it to the map. The event can be received before or after the subscribe result is processed, resulting in a race.

If I knew when the subscription actually we finished successfully, I could then wait to call "createSubscription" to start getting data, but I can't because I don't know anything about the status of the subscription internally.

I also don't really have a good suggestion about how to fix this either.

[Matthias247]

Hmm, that's some interesting points (that also show how difficult it is how to build a robust implementation of such a system).

I think you are even pointing out multiple issues here:

• What is actually not an issue is that multithreading or ordering problems inside the WampClient. Because it internally uses a single thread it can guarantee that all received messages will be processed in order.
• However there really seems to be a small data race here. If a subscribe message the received the action is delayed through rescheduling to the eventloop (https://github.com/Matthias247/jawampa/blob/master/src/main/java/ws/wamp/jawampa/WampClient.java#L1300). This has the advantage that it decouples the actions. However it might lead to problem you describe. If an Event message is pushed from the transport before the handler is executed then the event will be dropped. The chance for this is really minimal (probably only when both Subscribed and Event messages are inside the same IP frame), but nevertheless it exists. Removing .subscribeOn should remedy this, but it needs to be checked whether this creates any side effects. For most distributed applications it shouldn't be an issue anyway, because there's always the possiblity that that the server already sent events that are important to you before you connected and subscribed and you might want to design your API around that fact. However this really might require the information whether subscribe was successful.
• I agree that being able to retrieve the information that a subscribe succeeded would be helpful in some situations. But I'm currently also not sure what is the best way to express this in the API. One solution would be to have something like Observable<Observable<PubSubData>> makeSubscribption() where the first subscribed stream only yields a single value that signals that the subscription was succesful and the result is the actual stream of values to which you would need to subscribe. But the disadvantage is that this is very verbose, that the inner observable would need to buffer incoming events (the user would only get the chance to subscribe later to the Observable later than when the first events could be pushed) and that unsubscribing also gets tricky (Depending on the state you would need to either unsubscribe the outer or inner observable or even both).
  Another idea would be that the API stays as it is and the stream always outputs a special first element that only signals success for the subscription. Some kind of static marker instance of the PubSubData class. But for the overloaded variants of makeSubscription which return Observable<T> instead of Observable<PubSubData> it gets tricky again. I don't see a way to have a marker instance for every possible T type.

[jrogers]

Yes I meant a data race, I didn't mean to imply that there was a threading issue. In this particular scenario the services are running locally on the same host so that's probably why I'm seeing it more frequently, in a fully distributed setup the likelihood would seem to be low.

As for the changing the API or a potential fix, I agree I don't really like the observable of observable approach either. For now I've just forked locally and removed the subscribeOn to work around it, but I don't really think this is the correct fix for what I'm trying to do either. I guess I'll keep looking into a proper solution. I'm relatively new to RxJava as well, so I'm not sure if there's a pattern for this type of use case... On Jun 14, 2015 9:56 AM, "Matthias Einwag" notifications@github.com wrote:

Hmm, that's some interesting points (that also show how difficult it is how to build a robust implementation of such a system).

I think you are even pointing out multiple issues here:

• What is actually not an issue is that multithreading or ordering problems inside the WampClient. Because it internally uses a single thread it can guarantee that all received messages will be processed in order.
• However there really seems to be a small data race here. If a subscribe message the received the action is delayed through rescheduling to the eventloop ( https://github.com/Matthias247/jawampa/blob/master/src/main/java/ws/wamp/jawampa/WampClient.java#L1300). This has the advantage that it decouples the actions. However it might lead to problem you describe. If an Event message is pushed from the transport before the handler is executed then the event will be dropped. The chance for this is really minimal (probably only when both Subscribed and Event messages are inside the same IP frame), but nevertheless it exists. Removing .subscribeOn should remedy this, but it needs to be checked whether this creates any side effects. For most distributed applications it shouldn't be an issue anyway, because there's always the possiblity that that the serve r alread y sent events that are important to you before you connected and subscribed and you might want to design your API around that fact. However this really might require the information whether subscribe was successful.
• I agree that being able to retrieve the information that a subscribe succeeded would be helpful in some situations. But I'm currently also not sure what is the best way to express this in the API. One solution would be to have something like Observable<Observable> makeSubscribption() where the first subscribed stream only yields a single value that signals that the subscription was succesful and the result is the actual stream of values to which you would need to subscribe. But the disadvantage is that this is very verbose, that the inner observable would need to buffer incoming events (the user would only get the chance to subscribe later to the Observable later than when the first events could be pushed) and that unsubscribing also gets tricky (Depending on the state you would need to either unsubscribe the outer or inner observable or even both). Another idea would be that the API stays as it is and the stream always outputs a special first element that only signals success for the subscription. Some kind of static marker instance of the PubSubData class. But for the overloaded variants of makeSubscription which return Observable instead of Observable it gets tricky again. I don't see a way to have a marker instance for every possible T type.

— Reply to this email directly or view it on GitHub https://github.com/Matthias247/jawampa/issues/44#issuecomment-111828252.