Release ByteBuf after encoding to JSON

Byte buffers of Json objects backed by ByteBuf are now released after encoding.

[resolves #329]

Author: mp911de

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

@@ -18,7 +18,9 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
+import io.netty.util.ReferenceCountUtil;
 import io.r2dbc.postgresql.client.Parameter;
 import io.r2dbc.postgresql.message.Format;
 import io.r2dbc.postgresql.type.PostgresqlObjectId;
@@ -91,7 +93,11 @@ Parameter doEncode(Json value) {
             ByteBuf buffer = (ByteBuf) toEncode;
             buffer.touch();
 
-            return this.byteBufAllocator.buffer(buffer.readableBytes() + 1).writeByte(1).writeBytes(buffer);
+            try {
+                return this.byteBufAllocator.buffer(buffer.readableBytes() + 1).writeByte(1).writeBytes(buffer);
+            } finally {
+                ReferenceCountUtil.safeRelease(buffer);
+            }
         });
 
     }

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

@@ -16,13 +16,15 @@
 
 package io.r2dbc.postgresql.codec;
 
+import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.r2dbc.postgresql.client.Parameter;
 import io.r2dbc.postgresql.util.ByteBufUtils;
 import org.junit.jupiter.api.Test;
 
 import java.io.ByteArrayInputStream;
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 
 import static io.r2dbc.postgresql.client.Parameter.NULL_VALUE;
 import static io.r2dbc.postgresql.client.ParameterAssert.assertThat;
@@ -34,6 +36,7 @@
 import static io.r2dbc.postgresql.util.TestByteBufAllocator.TEST;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
 
 /**
  * Unit tests for {@link JsonCodec}.
@@ -110,6 +113,32 @@ void doEncode() {
             .hasValue(Unpooled.wrappedBuffer(Unpooled.wrappedBuffer(new byte[]{1}), ByteBufUtils.encode(TEST, json)));
     }
 
+    @Test
+    void doEncodeReleasedByteBuf() {
+        String json = "{\"name\":\"John Doe\"}";
+        JsonCodec jsonCodec = new JsonCodec(TEST);
+
+        ByteBuf buffer = TEST.buffer();
+        buffer.writeCharSequence(json, StandardCharsets.UTF_8);
+
+        assertThat(jsonCodec.doEncode(Json.of(buffer)))
+            .hasValue(Unpooled.wrappedBuffer(Unpooled.wrappedBuffer(new byte[]{1}), ByteBufUtils.encode(TEST, json)));
+
+        assertThat(buffer.refCnt()).isZero();
+    }
+
+    @Test
+    void doEncodeReleasedJsonOutput() {
+        String json = "{\"name\":\"John Doe\"}";
+        JsonCodec jsonCodec = new JsonCodec(TEST);
+        Json decodedBytes = jsonCodec.decode(ByteBufUtils.encode(TEST, json), JSON.getObjectId(), FORMAT_TEXT, Json.class);
+
+        assertThat(jsonCodec.doEncode(decodedBytes))
+            .hasValue(Unpooled.wrappedBuffer(Unpooled.wrappedBuffer(new byte[]{1}), ByteBufUtils.encode(TEST, json)));
+
+        assertThatIllegalStateException().isThrownBy(decodedBytes::asString).withMessage("JSON is already released");
+    }
+
     @Test
     void doEncodeNoValue() {
         assertThatIllegalArgumentException().isThrownBy(() -> new JsonCodec(TEST).doEncode(null))