Polishing.

Delay buffer allocation until there is something to read. Guard code with try/finally to ensure aggregation buffer release.

[pgjdbc#407][closes pgjdbc#408]

Signed-off-by: Mark Paluch <[email protected]>

Author: mp911de

src/main/java/io/r2dbc/postgresql/codec/HStoreCodec.java

@@ -25,7 +25,6 @@
 import reactor.core.publisher.Mono;
 
 import javax.annotation.Nullable;
-
 import java.nio.charset.StandardCharsets;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -37,25 +36,12 @@
 final class HStoreCodec implements Codec<Map> {
 
     /**
-     * A {@link ByteProcessor} which finds the first appearance of a specific byte.
+     * Aborts on a comma {@code ('"')}.
      */
-    static class IndexOfProcessor implements ByteProcessor {
-
-        private final byte byteToFind;
-
-        public IndexOfProcessor(byte byteToFind) {
-            this.byteToFind = byteToFind;
-        }
-
-        @Override
-        public boolean process(byte value) {
-            return value != this.byteToFind;
-        }
-
-    }
+    static final ByteProcessor FIND_QUOTE = new ByteProcessor.IndexOfProcessor((byte) '"');
 
     private static final Class<Map> TYPE = Map.class;
-    
+
     private final ByteBufAllocator byteBufAllocator;
 
     private final int oid;
@@ -163,9 +149,8 @@ private static byte peekByte(ByteBuf buffer) {
 
     @Nullable
     private static String readString(ByteBuf buffer) {
-        final ByteBuf accBuffer = buffer.alloc().buffer();
-        final int position = buffer.forEachByte(new IndexOfProcessor((byte) '"'));
 
+        int position = buffer.forEachByte(FIND_QUOTE);
         if (position > buffer.writerIndex()) {
             return null;
         }
@@ -174,20 +159,22 @@ private static String readString(ByteBuf buffer) {
             buffer.readerIndex(position + 1);
         }
 
-        while (buffer.isReadable()) {
-            final byte current = buffer.readByte();
-            
-            if (current == '"') {
-                break;
-            }
+        ByteBuf aggregated = buffer.alloc().buffer();
+        try {
+            while (buffer.isReadable()) {
+                byte current = buffer.readByte();
 
-            accBuffer.writeByte(current == '\\' ? buffer.readByte() : current);
-        }
+                if (current == '"') {
+                    break;
+                }
 
-        final String result = accBuffer.toString(StandardCharsets.UTF_8);
+                aggregated.writeByte(current == '\\' ? buffer.readByte() : current);
+            }
 
-        accBuffer.release();
-        return result;
+            return aggregated.toString(StandardCharsets.UTF_8);
+        } finally {
+            aggregated.release();
+        }
     }
 
     @Override

src/test/java/io/r2dbc/postgresql/codec/HStoreCodecUnitTests.java

@@ -20,7 +20,6 @@
 import io.r2dbc.postgresql.client.EncodedParameter;
 import io.r2dbc.postgresql.client.ParameterAssert;
 import io.r2dbc.postgresql.message.Format;
-import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import java.nio.charset.Charset;
@@ -62,7 +61,7 @@ void decodeAsText() {
         expect.put("d", "Zoë");
         expect.put("é", "120°");
 
-        Assertions.assertThat(res).isEqualTo(expect);
+        assertThat(res).isEqualTo(expect);
     }
 
     @Test