WebSockets are still connected after disconnect() is called

[stairs]

I'm facing the same problem which is described here. The condition for sockets being closed is never met as there is no promise that emitter is already canceled before doOnCancell() starts:

Flowable<String> flowable = Flowable.<String>create(mMessagesEmitters::add, BackpressureStrategy.BUFFER)
                .doOnCancel(() -> {
                    Iterator<FlowableEmitter<? super String>> iterator = mMessagesEmitters.iterator();
                    while (iterator.hasNext()) {
                        if (iterator.next().isCancelled()) iterator.remove();
                    }

                    if (mMessagesEmitters.size() < 1) {
                        Log.d(TAG, "Close web socket connection now in thread " + Thread.currentThread());
                        openedSocked.close(1000, "");
                        openedSocked = null;
                    }
                });

Here is a link for this code. Proof from RxJava2 repository.

[forresthopkinsa]

Yeah, there's a lot of really hacky Rx code in this branch. Can you do a test and ensure that the socket is indeed staying connected even when there are no subscribers at all? Thanks.

[stairs]

Actually, I face the same problem all around the library. For example, when I try to unsubscribe from the topic it doesn't send unsubscribe message. It appears that doOnCancel is called when the subscriber is not yet marked as canceled and hence is not deleted from emitters. I'm not sure how to fix this correctly yet. Do you suggest adding a test to the lib?

[forresthopkinsa]

I have a downstream branch that contains rewritten Rx logic. More info on 87.