Release ByteBuf after encoding.

Tomasz-Marciniak · mp911de · commit 922997c81d65 · 2025-09-09T08:54:46.000+02:00
Signed-off-by: Tomasz Marciniak <tomasz.marciniak@gmail.com> [resolves #279][#298]
diff --git a/src/jmh/java/io/r2dbc/mssql/PooledBenchmarks.java b/src/jmh/java/io/r2dbc/mssql/PooledBenchmarks.java
@@ -61,8 +61,7 @@ public ConnectionHolder() {
 
             this.jdbc = extension.getDataSource();
 
-            MssqlConnectionConfiguration configuration =
-                MssqlConnectionConfiguration.builder().host(extension.getHost()).username(extension.getUsername()).password(extension.getPassword()).build();
+            MssqlConnectionConfiguration configuration = extension.configBuilder().build();
 
             MssqlConnectionFactory mssqlConnectionFactory = new MssqlConnectionFactory(configuration);
             ConnectionPoolConfiguration poolConfiguration = ConnectionPoolConfiguration.builder(mssqlConnectionFactory).maxSize(4).build();
@@ -90,6 +89,7 @@ public void simpleDirectJdbc(ConnectionHolder connectionHolder, Blackhole voodoo
     }
 
     @Benchmark
+    @Testable
     public void simpleDirectR2dbc(ConnectionHolder connectionHolder, Blackhole voodoo) {
 
         String optname = Mono.usingWhen(connectionHolder.r2dbc.create(), connection -> {
diff --git a/src/main/java/io/r2dbc/mssql/codec/BlobCodec.java b/src/main/java/io/r2dbc/mssql/codec/BlobCodec.java
@@ -166,9 +166,14 @@ public Publisher<ByteBuffer> stream() {
                 result.flip();
                 return result;
             })
-                .doOnDiscard(ByteBuf.class, ByteBuf::release)
+                .doOnDiscard(ByteBuf.class, buffer ->
+                    {
+                        if (buffer.refCnt() > 0) {
+                            buffer.release();
+                        }
+                    }
+                )
                 .doOnCancel(() -> {
-
                     for (ByteBuf buffer : this.buffers) {
                         if (buffer.refCnt() > 0) {
                             buffer.release();
diff --git a/src/main/java/io/r2dbc/mssql/codec/ByteArray.java b/src/main/java/io/r2dbc/mssql/codec/ByteArray.java
@@ -42,10 +42,14 @@ static byte[] fromEncoded(Function<ByteBufAllocator, Encoded> encodeFunction) {
 
         Encoded encoded = encodeFunction.apply(ByteBufAllocator.DEFAULT);
 
+        ByteBuf buffer = null;
         try {
-            return ByteBufUtil.getBytes(encoded.getValue());
+            buffer = encoded.getValue();
+            return ByteBufUtil.getBytes(buffer);
         } finally {
-            encoded.dispose();
+            if (buffer != null && buffer.refCnt() > 0) {
+                buffer.release();
+            }
         }
     }
 
diff --git a/src/main/java/io/r2dbc/mssql/message/token/Attention.java b/src/main/java/io/r2dbc/mssql/message/token/Attention.java
@@ -79,6 +79,10 @@ public TdsFragment encode(ByteBufAllocator allocator, int packetSize) {
         ByteBuf buffer = allocator.buffer(length);
         encode(buffer);
 
+        if(buffer.refCnt() > 0) {
+            buffer.release();
+        }
+
         return TdsPackets.create(header, Unpooled.EMPTY_BUFFER);
     }
 
diff --git a/src/test/java/io/r2dbc/mssql/message/token/RowTokenUnitTests.java b/src/test/java/io/r2dbc/mssql/message/token/RowTokenUnitTests.java
@@ -67,6 +67,7 @@ void shouldDecodeRow() {
         assertThat(rowToken.getColumnData(1)).isNotNull();
         assertThat(rowToken.getColumnData(2)).isNotNull();
         assertThat(rowToken.getColumnData(3)).isNotNull();
+        buffer.release();
     }
 
     @Test
@@ -111,6 +112,9 @@ void shouldDecodeIntAndVarcharMax() throws IOException {
 
         assertThat(row.getColumnData(0).readableBytes()).isEqualTo(5);
         assertThat(row.getColumnData(1).readableBytes()).isEqualTo(10016);
+
+        rowData.release();
+        row.release();
     }
 
     @Test
@@ -129,6 +133,9 @@ void shouldDecodeIntAndVarcharMaxNull() {
 
         assertThat(row.getColumnData(0).readableBytes()).isEqualTo(5);
         assertThat(row.getColumnData(1)).isNull();
+
+        rowData.release();
+        row.release();
     }
 
     @Test
@@ -143,6 +150,7 @@ void canDecodeShouldReportPlpDecodability() throws IOException {
 
         ByteBuf rowData = loadRowData("int-varcharmax-data.txt");
         CanDecodeTestSupport.testCanDecode(rowData, buffer -> RowToken.canDecode(buffer, columns.getColumns()));
+        rowData.release();
     }
 
     @Test
diff --git a/src/test/java/io/r2dbc/mssql/util/EncodedAssert.java b/src/test/java/io/r2dbc/mssql/util/EncodedAssert.java
@@ -119,6 +119,10 @@ public EncodedAssert isEncodedAs(Consumer<ByteBuf> encoded) {
         Assertions.assertThat(ByteBufUtil.prettyHexDump(this.actual)).describedAs("ByteBuf")
             .isEqualTo(ByteBufUtil.prettyHexDump(expected));
 
+        if(this.actual.refCnt()>0){
+            this.actual.release();
+        }
+
         return this;
     }
 }
diff --git a/src/test/java/io/r2dbc/mssql/util/MsSqlServerExtension.java b/src/test/java/io/r2dbc/mssql/util/MsSqlServerExtension.java
@@ -90,7 +90,7 @@ public void afterAll(ExtensionContext context) {
     }
 
     public MssqlConnectionConfiguration.Builder configBuilder() {
-        return MssqlConnectionConfiguration.builder().host(getHost()).username(getUsername()).password(getPassword());
+        return MssqlConnectionConfiguration.builder().host(getHost()).port(getPort()).username(getUsername()).password(getPassword());
     }
 
     public MssqlConnectionConfiguration getConnectionConfiguration() {

Original file line number	Diff line number	Diff line change
`@@ -42,10 +42,14 @@ static byte[] fromEncoded(Function<ByteBufAllocator, Encoded> encodeFunction) {`
`42`	`42`
`43`	`43`	`Encoded encoded = encodeFunction.apply(ByteBufAllocator.DEFAULT);`
`44`	`44`
	`45`	`+ ByteBuf buffer = null;`
`45`	`46`	`try {`
`46`		`- return ByteBufUtil.getBytes(encoded.getValue());`
	`47`	`+ buffer = encoded.getValue();`
	`48`	`+ return ByteBufUtil.getBytes(buffer);`
`47`	`49`	`} finally {`
`48`		`- encoded.dispose();`
	`50`	`+ if (buffer != null && buffer.refCnt() > 0) {`
	`51`	`+ buffer.release();`
	`52`	`+ }`
`49`	`53`	`}`
`50`	`54`	`}`
`51`	`55`
Original file line number	Diff line number	Diff line change
`@@ -79,6 +79,10 @@ public TdsFragment encode(ByteBufAllocator allocator, int packetSize) {`
`79`	`79`	`ByteBuf buffer = allocator.buffer(length);`
`80`	`80`	`encode(buffer);`
`81`	`81`
	`82`	`+ if(buffer.refCnt() > 0) {`
	`83`	`+ buffer.release();`
	`84`	`+ }`
	`85`	`+`
`82`	`86`	`return TdsPackets.create(header, Unpooled.EMPTY_BUFFER);`
`83`	`87`	`}`
`84`	`88`
Original file line number	Diff line number	Diff line change
`@@ -119,6 +119,10 @@ public EncodedAssert isEncodedAs(Consumer<ByteBuf> encoded) {`
`119`	`119`	`Assertions.assertThat(ByteBufUtil.prettyHexDump(this.actual)).describedAs("ByteBuf")`
`120`	`120`	`.isEqualTo(ByteBufUtil.prettyHexDump(expected));`
`121`	`121`
	`122`	`+ if(this.actual.refCnt()>0){`
	`123`	`+ this.actual.release();`
	`124`	`+ }`
	`125`	`+`
`122`	`126`	`return this;`
`123`	`127`	`}`
`124`	`128`	`}`
Original file line number	Diff line number	Diff line change
`@@ -90,7 +90,7 @@ public void afterAll(ExtensionContext context) {`
`90`	`90`	`}`
`91`	`91`
`92`	`92`	`public MssqlConnectionConfiguration.Builder configBuilder() {`
`93`		`- return MssqlConnectionConfiguration.builder().host(getHost()).username(getUsername()).password(getPassword());`
	`93`	`+ return MssqlConnectionConfiguration.builder().host(getHost()).port(getPort()).username(getUsername()).password(getPassword());`
`94`	`94`	`}`
`95`	`95`
`96`	`96`	`public MssqlConnectionConfiguration getConnectionConfiguration() {`