Release bind parameters of unwritten messages.

[SecondPort]

Make sure that:

• You have read the contribution guidelines.
• You have created a feature request first to discuss your contribution intent. Please reference the feature request ticket number in the pull request.
• You use the code formatters provided here and have them applied to your changes. Don't submit any formatting related changes.
• You submit test cases (unit or integration tests) that back your changes.

Issue description

Resolves #734.

ExtendedFlowOperator.getMessages(…) retains the bound parameter buffers for the Bind it creates, and that reference is given back only by the finally block in Bind.encode(ByteBuf). ExtendedFlowOperator.close(…) returns the operator's own reference from doFinally. A buffer therefore reaches refCnt 0 only if the message is also encoded, and any message that never reaches the wire strands one reference per non-null parameter.

Two paths do that:

1. The conversation is refused before the request stream is subscribed. The three fetch* methods pass Flux.just(new CompositeFrontendMessage(factory.createMessages())) as an argument, so the messages — and their retains — are created while the exchange is assembled. When Client.exchange(…) refuses on !isConnected(), the request stream is never subscribed, doFinally(operator.close) returns one reference and the Bind's reference is stranded.

2. The channel dies while the message is queued. ChannelOperations.send(Publisher, Predicate) returns Mono.error(AbortedException.beforeSend()) on its !channel().isActive() branch without referencing dataStream. Since message.encode(alloc) is a cold Mono.fromSupplier, it is never subscribed, so Bind.encode(ByteBuf) and its finally never run. The sibling sendObject(Object) calls ReactorNetty.safeRelease(message) under the same check — reactor-netty releases what it holds by value, but it cannot release a cold publisher it never subscribed, so ownership stays with the caller.

This is the same class of request-side leak as the CopyData fix in 68694fc, one level up: FrontendMessage.dispose() already documents that a message discarded before reaching the wire must release its buffers and that disposal is idempotent. These two paths simply did not honour it.

A correction to the issue report, which I filed: a bare cancel over a live connection does not leak. Operators.discardOnCancel does not cancel upstream — it keeps draining until the conversation terminates, so doFinally(operator.close) fires late and Bind.encode's finally has already run by then. Measured: refCnt is still 1 immediately after thenCancel() and 0 only after the connection closes. The leak requires the message never to be encoded; cancellation is the common real-world trigger, because a cancelled chain releases or closes the connection. The issue title is imprecise on that point.

Also worth noting: Client.send(FrontendMessage) does not leak and is unchanged. doSendRequest wraps the message in Mono.just(…), so concatMap treats it as a scalar and WeakScalarSubscription.cancel() re-discards it into the existing doOnDiscard(FrontendMessage.class, FrontendMessage::dispose). The exchange(…) shape is a non-scalar FluxConcatArray, where an already-emitted value is not re-discarded.

New Public APIs

None. No signature changes, no new configuration.

Additional context

The change is +12/-4 across two files:

• ExtendedFlowDelegate — create the messages upon subscription (Mono.fromSupplier) instead of while assembling the exchange, at the three fetch* sites. Nothing is retained for a request stream that is never subscribed, which covers the exchange(…) refusal, addConversation's subscribe-time re-check and a full conversation queue in one place.
• ReactorNettyClient — dispose the message once the write attempt terminated. dispose() is idempotent and a no-op for messages that got encoded (Bind guards on its AtomicBoolean, CopyData on refCnt() > 0, CompositeFrontendMessage propagates, the default is a no-op).

I deliberately did not collapse the double ownership by dropping the extra retain() in getMessages(…): operator.close(…) fires from doFinally and can run while the Bind is still queued, which would hand released buffers to Bind.encode(ByteBuf).

Tests are unit tests rather than integration tests, because PR CI runs with -D skipITs and a regression test under *IntegrationTests would not guard this. Both new classes drive a real ReactorNettyClient over a loopback socket — no Docker, no PostgreSQL — and assert refCnt() directly rather than relying on the sampling leak detector.

• ReactorNettyClientUnitTests (the class removed in 86ab56d, reinstated) — path 2, plus a control test on a live channel.
• PostgresqlStatementCancellationUnitTests — both paths end to end through PostgresqlStatement, plus a control test that cancels over a live connection.

The control tests pass without the fix; they are what shows the failing assertions are the bug rather than a harness artefact. Verified on this branch: without the src/main changes the three regression tests fail with expected: 0 but was: 1 and the two control tests pass; with them, ./mvnw -B verify reports 1413 unit tests green.