Fix buffer leak during compression
Ensure that buffers used for uncompressed message prior to compression
are released.
@@ -264,24 +264,30 @@ public class InternalStreamConnection implements InternalConnection {
}
private void sendCommandMessage(final CommandMessage message,
final ByteBufferBsonOutput bsonOutput, final SessionContext sessionContext) {
- try {
if (sendCompressor == null || SECURITY_SENSITIVE_COMMANDS.contains(message.getCommandDocument(bsonOutput).getFirstKey())) {
+ try {
sendMessage(bsonOutput.getByteBuffers(), message.getId());
+ } finally {
+ bsonOutput.close();
+ }
} else {
- CompressedMessage compressedMessage = new CompressedMessage(message.getOpCode(), bsonOutput.getByteBuffers(),
- sendCompressor,
+ List<ByteBuf> byteBuffers = bsonOutput.getByteBuffers();
+ ByteBufferBsonOutput compressedBsonOutput;
+ try {
+ CompressedMessage compressedMessage = new CompressedMessage(message.getOpCode(), byteBuffers, sendCompressor,
getMessageSettings(description));
- ByteBufferBsonOutput compressedBsonOutput = new ByteBufferBsonOutput(this);
+ compressedBsonOutput = new ByteBufferBsonOutput(this);
compressedMessage.encode(compressedBsonOutput, sessionContext);
+ } finally {
+ releaseAllBuffers(byteBuffers);
+ bsonOutput.close();
+ }
try {
sendMessage(compressedBsonOutput.getByteBuffers(), message.getId());
} finally {
compressedBsonOutput.close();
}
}
- } finally {
- bsonOutput.close();
- }
}
private <T> T receiveCommandMessageResponse(final CommandMessage message, final Decoder<T> decoder,
@@ -324,11 +330,15 @@ public class InternalStreamConnection implements InternalConnection {
sendCommandMessageAsync(message.getId(), decoder, sessionContext, callback, bsonOutput, commandEventSender,
message.isResponseExpected());
} else {
- CompressedMessage compressedMessage = new CompressedMessage(message.getOpCode(), bsonOutput.getByteBuffers(),
- sendCompressor,
+ List<ByteBuf> byteBuffers = bsonOutput.getByteBuffers();
+ try {
+ CompressedMessage compressedMessage = new CompressedMessage(message.getOpCode(), byteBuffers, sendCompressor,
getMessageSettings(description));
compressedMessage.encode(compressedBsonOutput, sessionContext);
+ } finally {
+ releaseAllBuffers(byteBuffers);
bsonOutput.close();
+ }
sendCommandMessageAsync(message.getId(), decoder, sessionContext, callback, compressedBsonOutput, commandEventSender,
message.isResponseExpected());
}
@@ -339,6 +349,12 @@ public class InternalStreamConnection implements InternalConnection {
}
}
+ private void releaseAllBuffers(final List<ByteBuf> byteBuffers) {
+ for (ByteBuf cur : byteBuffers) {
+ cur.release();
+ }
+ }
+
private <T> void sendCommandMessageAsync(final int messageId, final Decoder<T> decoder, final SessionContext sessionContext,
final SingleResultCallback<T> callback, final ByteBufferBsonOutput bsonOutput,
final CommandEventSender commandEventSender, final boolean responseExpected) {
