Enable BlobCodec and ClobCodec for binary usage

[resolves #186]

Author: mp911de

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

@@ -57,14 +57,14 @@ boolean doCanDecode(PostgresqlObjectId type, Format format) {
         Assert.requireNonNull(format, "format must not be null");
         Assert.requireNonNull(type, "type must not be null");
 
-        return FORMAT_TEXT == format && BYTEA == type;
+        return BYTEA == type;
     }
 
     @Override
     Blob doDecode(ByteBuf buffer, PostgresqlObjectId dataType, @Nullable Format format, @Nullable Class<? extends Blob> type) {
         Assert.requireNonNull(buffer, "byteBuf must not be null");
 
-        return new ByteABlob(buffer);
+        return new ByteABlob(buffer, format);
     }
 
     @Override
@@ -103,29 +103,37 @@ private static final class ByteABlob implements Blob {
 
         private final ByteBuf byteBuf;
 
-        private ByteABlob(ByteBuf byteBuf) {
+        private final Format format;
+
+        private ByteABlob(ByteBuf byteBuf, Format format) {
             this.byteBuf = byteBuf.retain();
+            this.format = format;
         }
 
         @Override
         public Mono<Void> discard() {
-            return Mono.defer(() -> {
-                this.byteBuf.release();
-                return Mono.empty();
+            return Mono.fromRunnable(() -> {
+                if (this.byteBuf.refCnt() > 0) {
+                    this.byteBuf.release();
+                }
             });
         }
 
         @Override
         public Mono<ByteBuffer> stream() {
-            return Mono.defer(() -> {
+            return Mono.fromSupplier(() -> {
+                if (this.format == Format.FORMAT_BINARY) {
+                    return this.byteBuf.nioBuffer();
+                }
+
                 Matcher matcher = BLOB_PATTERN.matcher(ByteBufUtils.decode(this.byteBuf));
 
                 if (!matcher.find()) {
-                    return Mono.error(new IllegalStateException("ByteBuf does not contain BYTEA hex format"));
+                    throw new IllegalStateException("ByteBuf does not contain BYTEA hex format");
                 }
 
-                return Mono.just(ByteBuffer.wrap(ByteBufUtil.decodeHexDump(matcher.group(1))));
-            });
+                return ByteBuffer.wrap(ByteBufUtil.decodeHexDump(matcher.group(1)));
+            }).doAfterTerminate(this.byteBuf::release);
         }
     }
 

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

@@ -51,7 +51,7 @@ boolean doCanDecode(PostgresqlObjectId type, Format format) {
         Assert.requireNonNull(format, "format must not be null");
         Assert.requireNonNull(type, "type must not be null");
 
-        return FORMAT_TEXT == format && TEXT == type;
+        return TEXT == type;
     }
 
     @Override

src/test/java/io/r2dbc/postgresql/AbstractCodecIntegrationTests.java

@@ -22,10 +22,14 @@
 import io.r2dbc.postgresql.api.PostgresqlResult;
 import io.r2dbc.postgresql.api.PostgresqlStatement;
 import io.r2dbc.postgresql.codec.Json;
+import io.r2dbc.spi.Blob;
+import io.r2dbc.spi.Clob;
 import io.r2dbc.spi.Connection;
 import org.assertj.core.data.Offset;
 import org.junit.jupiter.api.Test;
+import org.reactivestreams.Publisher;
 import org.springframework.util.StreamUtils;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
@@ -52,6 +56,7 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import static io.r2dbc.postgresql.util.TestByteBufAllocator.TEST;
 import static org.assertj.core.api.Assertions.assertThat;
 
 abstract class AbstractCodecIntegrationTests extends AbstractIntegrationTests {
@@ -92,6 +97,57 @@ void charPrimitive() {
         testCodec(Character.class, 'a', "VARCHAR(1)");
     }
 
+    @Test
+    void blob() {
+        byte[] bytes = {1, 2, 3, 4};
+        testCodec(Blob.class,
+            new Blob() {
+
+                @Override
+                public Publisher<Void> discard() {
+                    return Mono.empty();
+                }
+
+                @Override
+                public Publisher<ByteBuffer> stream() {
+                    return Mono.just(ByteBuffer.wrap(bytes));
+                }
+            },
+            (actual, expected) -> Flux.zip(
+                Flux.from(actual.stream()).reduce(TEST.heapBuffer(), ByteBuf::writeBytes),
+                Flux.from(expected.stream()).reduce(TEST.heapBuffer(), ByteBuf::writeBytes)
+            )
+                .as(StepVerifier::create)
+                .assertNext(t -> assertThat(t.getT1()).isEqualTo(t.getT2()))
+                .verifyComplete()
+            , "BYTEA");
+    }
+
+    @Test
+    void clob() {
+        testCodec(Clob.class,
+            new Clob() {
+
+                @Override
+                public Publisher<Void> discard() {
+                    return Mono.empty();
+                }
+
+                @Override
+                public Publisher<CharSequence> stream() {
+                    return Mono.just("test-value");
+                }
+            },
+            (actual, expected) -> Flux.zip(
+                Flux.from(actual.stream()).reduce(new StringBuilder(), StringBuilder::append).map(StringBuilder::toString),
+                Flux.from(expected.stream()).reduce(new StringBuilder(), StringBuilder::append).map(StringBuilder::toString)
+            )
+                .as(StepVerifier::create)
+                .assertNext(t -> assertThat(t.getT1()).isEqualToIgnoringWhitespace(t.getT2()))
+                .verifyComplete()
+            , "TEXT");
+    }
+
     @Test
     void date() {
         testCodec(Date.class, new Date(), "TIMESTAMP");