error during connection to pub/sub is not transformed to stream failure

[aludwiko]

I noticed one thing. When I'm trying to connect to not existing subscription (or not yet created) stream is not failing. It hangs and it's not possible to capture the error in any way. On debugger I can see that this part of the code is invoked:

queue.put(Left(InternalPubSubError(failure)))

but somehow takeNextElement method is not returning anything.

[aludwiko]

After some more debugging, this line will never succeed:

queue.put(Left(InternalPubSubError(failure)))

because of InterruptedException during putLock.lockInterruptibly:

    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        final int c;
        final Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();

[bastewart]

Ah, thanks for the report. @TimWSpence was looking to replace the queue implementation for performance reasons (#372) so this is definitely something we could/should fix at the same time.

[fugafree]

Hello @bastewart, may I ask what is the status of this? It is causing serious issues for us, as broken connections could stay unnoticed.

[alejandrohdezma]

This issue should not be appearing on the new library fs2-pubsub but feel free to re-open if that is still the case and I'll investigate.