Polishing

Reorder methods according to interface declaration. Add since tags.

[pgjdbc#230]

Author: mp911de

src/main/java/io/r2dbc/postgresql/client/Client.java

@@ -68,6 +68,7 @@ default Flux<BackendMessage> exchange(Publisher<FrontendMessage> requests) {
      * @param requests the publisher of outbound messages
      * @return a {@link Flux} of incoming messages that ends with the end of the frame (i.e. reception of a {@link ReadyForQuery} message.
      * @throws IllegalArgumentException if {@code requests} is {@code null}
+     * @since 0.9
      */
     Flux<BackendMessage> exchange(Predicate<BackendMessage> takeUntil, Publisher<FrontendMessage> requests);
 
@@ -76,6 +77,7 @@ default Flux<BackendMessage> exchange(Publisher<FrontendMessage> requests) {
      *
      * @param message outbound message
      * @throws IllegalArgumentException if {@code message} is {@code null}
+     * @since 0.9
      */
     void send(FrontendMessage message);
 

src/main/java/io/r2dbc/postgresql/client/ReactorNettyClient.java

@@ -172,6 +172,35 @@ private ReactorNettyClient(Connection connection) {
             .subscribe();
     }
 
+    @Override
+    public Disposable addNotificationListener(Consumer<NotificationResponse> consumer) {
+        return this.notificationProcessor.subscribe(consumer);
+    }
+
+    @Override
+    public Mono<Void> close() {
+        return Mono.defer(() -> {
+
+            drainError(EXPECTED);
+            if (this.isClosed.compareAndSet(false, true)) {
+
+                if (!isConnected() || this.processId == null) {
+                    this.connection.dispose();
+                    return this.connection.onDispose();
+                }
+
+                return Flux.just(Terminate.INSTANCE)
+                    .doOnNext(message -> logger.debug("Request:  {}", message))
+                    .concatMap(message -> this.connection.outbound().send(message.encode(this.connection.outbound().alloc())))
+                    .then()
+                    .doOnSuccess(v -> this.connection.dispose())
+                    .then(this.connection.onDispose());
+            }
+
+            return Mono.empty();
+        });
+    }
+
     @Override
     public Flux<BackendMessage> exchange(Predicate<BackendMessage> takeUntil, Publisher<FrontendMessage> requests) {
         Assert.requireNonNull(takeUntil, "takeUntil must not be null");
@@ -355,38 +384,6 @@ private static Mono<? extends Void> registerSslHandler(SSLConfig sslConfig, Conn
         return Mono.empty();
     }
 
-    @Override
-    public Mono<Void> close() {
-        return Mono.defer(() -> {
-
-            drainError(EXPECTED);
-            if (this.isClosed.compareAndSet(false, true)) {
-
-                if (!isConnected() || this.processId == null) {
-                    this.connection.dispose();
-                    return this.connection.onDispose();
-                }
-
-                return Flux.just(Terminate.INSTANCE)
-                    .doOnNext(message -> logger.debug("Request:  {}", message))
-                    .concatMap(message -> this.connection.outbound().send(message.encode(this.connection.outbound().alloc())))
-                    .then()
-                    .doOnSuccess(v -> this.connection.dispose())
-                    .then(this.connection.onDispose());
-            }
-
-            return Mono.empty();
-        });
-    }
-
-    private void drainError(Supplier<? extends Throwable> supplier) {
-        Conversation receiver;
-
-        while ((receiver = this.conversations.poll()) != null) {
-            receiver.sink.error(supplier.get());
-        }
-    }
-
     @Override
     public ByteBufAllocator getByteBufAllocator() {
         return this.byteBufAllocator;
@@ -426,11 +423,6 @@ public boolean isConnected() {
         return channel.isOpen();
     }
 
-    @Override
-    public Disposable addNotificationListener(Consumer<NotificationResponse> consumer) {
-        return this.notificationProcessor.subscribe(consumer);
-    }
-
     private static String toString(List<Field> fields) {
 
         StringJoiner joiner = new StringJoiner(", ");
@@ -453,6 +445,14 @@ private void handleConnectionError(Throwable error) {
         drainError(() -> new PostgresConnectionException(error));
     }
 
+    private void drainError(Supplier<? extends Throwable> supplier) {
+        Conversation receiver;
+
+        while ((receiver = this.conversations.poll()) != null) {
+            receiver.sink.error(supplier.get());
+        }
+    }
+
     /**
      * Value object representing a single conversation. The driver permits a single conversation at a time to ensure that request messages get routed to the proper response receiver and do not leak
      * into other conversations. A conversation must be finished in the sense that the {@link Publisher} of {@link FrontendMessage} has completed before the next conversation is started.