@@ -264,23 +264,29 @@ public <T> T sendAndReceive(final CommandMessage message, final Decoder<T> decod
264
264
}
265
265
private void sendCommandMessage (final CommandMessage message ,
266
266
final ByteBufferBsonOutput bsonOutput , final SessionContext sessionContext ) {
267
- try {
268
- if ( sendCompressor == null || SECURITY_SENSITIVE_COMMANDS . contains ( message . getCommandDocument ( bsonOutput ). getFirstKey ())) {
267
+ if ( sendCompressor == null || SECURITY_SENSITIVE_COMMANDS . contains ( message . getCommandDocument ( bsonOutput ). getFirstKey ())) {
268
+ try {
269
269
sendMessage (bsonOutput .getByteBuffers (), message .getId ());
270
- } else {
271
- CompressedMessage compressedMessage = new CompressedMessage (message .getOpCode (), bsonOutput .getByteBuffers (),
272
- sendCompressor ,
270
+ } finally {
271
+ bsonOutput .close ();
272
+ }
273
+ } else {
274
+ List <ByteBuf > byteBuffers = bsonOutput .getByteBuffers ();
275
+ ByteBufferBsonOutput compressedBsonOutput ;
276
+ try {
277
+ CompressedMessage compressedMessage = new CompressedMessage (message .getOpCode (), byteBuffers , sendCompressor ,
273
278
getMessageSettings (description ));
274
- ByteBufferBsonOutput compressedBsonOutput = new ByteBufferBsonOutput (this );
279
+ compressedBsonOutput = new ByteBufferBsonOutput (this );
275
280
compressedMessage .encode (compressedBsonOutput , sessionContext );
276
- try {
277
- sendMessage (compressedBsonOutput .getByteBuffers (), message .getId ());
278
- } finally {
279
- compressedBsonOutput .close ();
280
- }
281
+ } finally {
282
+ releaseAllBuffers (byteBuffers );
283
+ bsonOutput .close ();
284
+ }
285
+ try {
286
+ sendMessage (compressedBsonOutput .getByteBuffers (), message .getId ());
287
+ } finally {
288
+ compressedBsonOutput .close ();
281
289
}
282
- } finally {
283
- bsonOutput .close ();
284
290
}
285
291
}
286
292
@@ -324,11 +330,15 @@ public <T> void sendAndReceiveAsync(final CommandMessage message, final Decoder<
324
330
sendCommandMessageAsync (message .getId (), decoder , sessionContext , callback , bsonOutput , commandEventSender ,
325
331
message .isResponseExpected ());
326
332
} else {
327
- CompressedMessage compressedMessage = new CompressedMessage (message .getOpCode (), bsonOutput .getByteBuffers (),
328
- sendCompressor ,
329
- getMessageSettings (description ));
330
- compressedMessage .encode (compressedBsonOutput , sessionContext );
331
- bsonOutput .close ();
333
+ List <ByteBuf > byteBuffers = bsonOutput .getByteBuffers ();
334
+ try {
335
+ CompressedMessage compressedMessage = new CompressedMessage (message .getOpCode (), byteBuffers , sendCompressor ,
336
+ getMessageSettings (description ));
337
+ compressedMessage .encode (compressedBsonOutput , sessionContext );
338
+ } finally {
339
+ releaseAllBuffers (byteBuffers );
340
+ bsonOutput .close ();
341
+ }
332
342
sendCommandMessageAsync (message .getId (), decoder , sessionContext , callback , compressedBsonOutput , commandEventSender ,
333
343
message .isResponseExpected ());
334
344
}
@@ -339,6 +349,12 @@ public <T> void sendAndReceiveAsync(final CommandMessage message, final Decoder<
339
349
}
340
350
}
341
351
352
+ private void releaseAllBuffers (final List <ByteBuf > byteBuffers ) {
353
+ for (ByteBuf cur : byteBuffers ) {
354
+ cur .release ();
355
+ }
356
+ }
357
+
342
358
private <T > void sendCommandMessageAsync (final int messageId , final Decoder <T > decoder , final SessionContext sessionContext ,
343
359
final SingleResultCallback <T > callback , final ByteBufferBsonOutput bsonOutput ,
344
360
final CommandEventSender commandEventSender , final boolean responseExpected ) {
0 commit comments