Remove redundant lambda expression (#125)

**Motivation:**

See #124 for more.

**Modification:**

Remove redundant static variables which are used only once and use
method reference instead to improve code readability.

**Result:**

Closes #124

Author: JohnNiang

src/main/java/io/asyncer/r2dbc/mysql/MySqlConnection.java

@@ -76,36 +76,6 @@ public final class MySqlConnection implements Connection, ConnectionState {
 
     private static final ServerVersion TX_LEVEL_8X = ServerVersion.create(8, 0, 0);
 
-    /**
-     * Convert initialize result to {@link InitData}.
-     */
-    private static final Function<MySqlResult, Publisher<InitData>> INIT_HANDLER = r ->
-        r.map((row, meta) -> new InitData(convertIsolationLevel(row.get(0, String.class)),
-            convertLockWaitTimeout(row.get(1, Long.class)),
-            row.get(2, String.class), null));
-
-    private static final Function<MySqlResult, Publisher<InitData>> FULL_INIT = r -> r.map((row, meta) -> {
-        IsolationLevel level = convertIsolationLevel(row.get(0, String.class));
-        long lockWaitTimeout = convertLockWaitTimeout(row.get(1, Long.class));
-        String product = row.get(2, String.class);
-        String systemTimeZone = row.get(3, String.class);
-        String timeZone = row.get(4, String.class);
-        ZoneId zoneId;
-
-        if (timeZone == null || timeZone.isEmpty() || "SYSTEM".equalsIgnoreCase(timeZone)) {
-            if (systemTimeZone == null || systemTimeZone.isEmpty()) {
-                logger.warn("MySQL does not return any timezone, trying to use system default timezone");
-                zoneId = ZoneId.systemDefault();
-            } else {
-                zoneId = convertZoneId(systemTimeZone);
-            }
-        } else {
-            zoneId = convertZoneId(timeZone);
-        }
-
-        return new InitData(level, lockWaitTimeout, product, zoneId);
-    });
-
     private static final BiConsumer<ServerMessage, SynchronousSink<Boolean>> PING = (message, sink) -> {
         if (message instanceof ErrorMessage) {
             ErrorMessage msg = (ErrorMessage) message;
@@ -456,9 +426,9 @@ static Mono<MySqlConnection> init(Client client, Codecs codecs, ConnectionContex
 
         if (context.shouldSetServerZoneId()) {
             query.append(",@@system_time_zone AS s,@@time_zone AS t");
-            handler = FULL_INIT;
+            handler = MySqlConnection::fullInit;
         } else {
-            handler = INIT_HANDLER;
+            handler = MySqlConnection::init;
         }
 
         return new TextSimpleStatement(client, codecs, context, query.toString())
@@ -477,6 +447,36 @@ static Mono<MySqlConnection> init(Client client, Codecs codecs, ConnectionContex
             });
     }
 
+    private static Publisher<InitData> init(MySqlResult r) {
+        return r.map((row, meta) -> new InitData(convertIsolationLevel(row.get(0, String.class)),
+            convertLockWaitTimeout(row.get(1, Long.class)),
+            row.get(2, String.class), null));
+    }
+
+    private static Publisher<InitData> fullInit(MySqlResult r) {
+        return r.map((row, meta) -> {
+            IsolationLevel level = convertIsolationLevel(row.get(0, String.class));
+            long lockWaitTimeout = convertLockWaitTimeout(row.get(1, Long.class));
+            String product = row.get(2, String.class);
+            String systemTimeZone = row.get(3, String.class);
+            String timeZone = row.get(4, String.class);
+            ZoneId zoneId;
+
+            if (timeZone == null || timeZone.isEmpty() || "SYSTEM".equalsIgnoreCase(timeZone)) {
+                if (systemTimeZone == null || systemTimeZone.isEmpty()) {
+                    logger.warn("MySQL does not return any timezone, trying to use system default timezone");
+                    zoneId = ZoneId.systemDefault();
+                } else {
+                    zoneId = convertZoneId(systemTimeZone);
+                }
+            } else {
+                zoneId = convertZoneId(timeZone);
+            }
+
+            return new InitData(level, lockWaitTimeout, product, zoneId);
+        });
+    }
+
     /**
      * Creates a {@link ZoneId} from MySQL timezone result, or fallback to system default timezone if not
      * found.

src/main/java/io/asyncer/r2dbc/mysql/MySqlConnectionConfiguration.java

@@ -47,8 +47,6 @@ public final class MySqlConnectionConfiguration {
      */
     private static final int DEFAULT_PORT = 3306;
 
-    private static final Predicate<String> DEFAULT_SERVER_PREPARE = sql -> false;
-
     /**
      * {@code true} if {@link #domain} is hostname, otherwise {@link #domain} is unix domain socket path.
      */
@@ -685,7 +683,7 @@ public Builder useClientPrepareStatement() {
          * @since 0.8.1
          */
         public Builder useServerPrepareStatement() {
-            return useServerPrepareStatement(DEFAULT_SERVER_PREPARE);
+            return useServerPrepareStatement((sql) -> false);
         }
 
         /**

src/main/java/io/asyncer/r2dbc/mysql/MySqlResult.java

@@ -42,7 +42,6 @@
 
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
-import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
 
@@ -57,20 +56,6 @@
  */
 public final class MySqlResult implements Result {
 
-    private static final Consumer<ReferenceCounted> RELEASE = ReferenceCounted::release;
-
-    private static final BiConsumer<Segment, SynchronousSink<Long>> ROWS_UPDATED = (segment, sink) -> {
-        if (segment instanceof UpdateCount) {
-            sink.next(((UpdateCount) segment).value());
-        } else if (segment instanceof Message) {
-            sink.error(((Message) segment).exception());
-        } else if (segment instanceof ReferenceCounted) {
-            ReferenceCountUtil.safeRelease(segment);
-        }
-    };
-
-    private static final BiFunction<Long, Long, Long> SUM = Long::sum;
-
     private final Flux<Segment> segments;
 
     private MySqlResult(Flux<Segment> segments) {
@@ -79,7 +64,15 @@ private MySqlResult(Flux<Segment> segments) {
 
     @Override
     public Mono<Long> getRowsUpdated() {
-        return segments.handle(ROWS_UPDATED).reduce(SUM);
+        return segments.<Long>handle((segment, sink) -> {
+            if (segment instanceof UpdateCount) {
+                sink.next(((UpdateCount) segment).value());
+            } else if (segment instanceof Message) {
+                sink.error(((Message) segment).exception());
+            } else if (segment instanceof ReferenceCounted) {
+                ReferenceCountUtil.safeRelease(segment);
+            }
+        }).reduce(Long::sum);
     }
 
     @Override
@@ -168,7 +161,7 @@ static MySqlResult toResult(boolean binary, Codecs codecs, ConnectionContext con
         requireNonNull(messages, "messages must not be null");
 
         return new MySqlResult(OperatorUtils.discardOnCancel(messages)
-            .doOnDiscard(ReferenceCounted.class, RELEASE)
+            .doOnDiscard(ReferenceCounted.class, ReferenceCounted::release)
             .handle(new MySqlSegments(binary, codecs, context, generatedKeyName)));
     }
 
@@ -220,10 +213,6 @@ public Row row() {
 
         @Override
         public ReferenceCounted touch(Object hint) {
-            if (this.fields.length == 0) {
-                return this;
-            }
-
             for (FieldValue field : this.fields) {
                 field.touch(hint);
             }

src/main/java/io/asyncer/r2dbc/mysql/client/ReactorNettyClient.java

@@ -46,7 +46,6 @@
 
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiConsumer;
-import java.util.function.Consumer;
 import java.util.function.Function;
 
 import static io.asyncer.r2dbc.mysql.internal.util.AssertUtils.require;
@@ -63,8 +62,6 @@ final class ReactorNettyClient implements Client {
 
     private static final boolean INFO_ENABLED = logger.isInfoEnabled();
 
-    private static final Consumer<ReferenceCounted> RELEASE = ReferenceCounted::release;
-
     private final Connection connection;
 
     private final ConnectionContext context;
@@ -153,7 +150,7 @@ public <T> Flux<T> exchange(ClientMessage request,
                                                                       .doOnSubscribe(ignored -> emitNextRequest(request))
                                                                       .handle(handler)
                                                                       .doOnTerminate(requestQueue))
-                                             .doOnDiscard(ReferenceCounted.class, RELEASE);
+                                             .doOnDiscard(ReferenceCounted.class, ReferenceCounted::release);
 
             requestQueue.submit(RequestTask.wrap(request, sink, responses));
         }).flatMapMany(Function.identity());
@@ -187,7 +184,7 @@ public <T> Flux<T> exchange(FluxExchangeable<T> exchangeable) {
                     });
 
             requestQueue.submit(RequestTask.wrap(exchangeable, sink, OperatorUtils.discardOnCancel(responses)
-                .doOnDiscard(ReferenceCounted.class, RELEASE)
+                .doOnDiscard(ReferenceCounted.class, ReferenceCounted::release)
                 .doOnCancel(exchangeable::dispose)));
         }).flatMapMany(Function.identity());
     }

src/main/java/io/asyncer/r2dbc/mysql/message/client/PreparedExecuteMessage.java

@@ -27,7 +27,6 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-import java.util.function.Consumer;
 
 import static io.asyncer.r2dbc.mysql.internal.util.AssertUtils.requireNonNull;
 
@@ -59,8 +58,6 @@ public final class PreparedExecuteMessage implements ClientMessage, Disposable {
 
     private static final byte EXECUTE_FLAG = 0x17;
 
-    private static final Consumer<MySqlParameter> DISPOSE = MySqlParameter::dispose;
-
     private final int statementId;
 
     /**
@@ -131,7 +128,7 @@ public Flux<ByteBuf> encode(ByteBufAllocator allocator, ConnectionContext contex
                 writeTypes(buf, size);
 
                 Flux<ByteBuf> parameters = OperatorUtils.discardOnCancel(Flux.fromArray(values))
-                    .doOnDiscard(MySqlParameter.class, DISPOSE)
+                    .doOnDiscard(MySqlParameter.class, MySqlParameter::dispose)
                     .concatMap(MySqlParameter::publishBinary);
 
                 return Flux.just(buf).concatWith(parameters);