Remove mutable state (connectionRef) from Transport class.

[taer]

Forward port the fix in #326 to master

[mikkokar]

Hi @taer

The observable should be constructed like this:

            public Observable<LiveHttpResponse> response() {
                return connection.flatMap(connection -> {
                    Observable<LiveHttpResponse> responseObservable = connection.write(request)
                            .map(response -> addOriginId(originId, response));

                    return ResponseEventListener.from(responseObservable)
                            .whenCancelled(() -> closeIfConnected(origin, connection))
                            .whenResponseError(cause -> closeIfConnected(origin, connection))
                            .whenContentError(cause -> closeIfConnected(origin, connection))
                            .whenCompleted(() -> returnIfConnected(origin, connection))
                            .apply();
                });
            }

The ResponseEventListener.from takes a response observable and decorates it with event listeners triggering at relevant points of the response lifecycle. It takes into account both response observable and its associated body byte stream. The apply method returns a new Observable<LiveHttpResponse> that should be returned for the downstream observers.

Also, I have pushed this to your branch. Please pull the lates changes to have a look!

[mikkokar]

Hi @taer. This is a good PR even if it doesn't actually fix the leak. Therefore I am going to rename the PR title accordingly and merge.

For the record, the root cause for the connection leak is lack of cancel/unsubscribehandling for connection observables. More specifically, Transport class handles cancellations for response observables, which only works after a connection event is published from the connection observable. The cancellation therefore doesn't have any effect when subscriber (connection) is pending.

As ultimate producers of connections, unsubscriptions must be dealt with in the connection pool, and in NettyConnectionFactory.