MsgStreamer extends OutputStream implements ObjToByteArraySerializer, BaseMsgStreamer, ByteBufferWriter { @Override public int writeMessage() throws IOException { try { this.startedSerializingMsg = true; InternalDataSerializer.writeDSFID(this.msg, this); this.doneWritingMsg = true; if (this.flushedBytes == 0) { this.normalMsg = true; } realFlush(true); return this.flushedBytes; } finally { release(); } } MsgStreamer(List<?> cons, DistributionMessage msg, boolean directReply, DMStats stats,
      int sendBufferSize, BufferPool bufferPool); @Override ConnectExceptions getConnectExceptions(); @Override List<?> getSentConnections(); static BaseMsgStreamer create(List<?> cons, final DistributionMessage msg,
      final boolean directReply, final DMStats stats, BufferPool bufferPool); @Override void reserveConnections(long startTime, long ackTimeout, long ackSDTimeout); @Override int writeMessage(); @Override void write(int b); @Override void flush(); void realFlush(boolean lastFlushForMessage); @Override void close(); @Override void write(byte[] source, int offset, int len); @Override void write(ByteBuffer bb); @Override void writeBoolean(boolean v); @Override void writeByte(int v); @Override void writeShort(int v); @Override void writeChar(int v); @Override void writeInt(int v); @Override void writeLong(long v); @Override void writeFloat(float v); @Override void writeDouble(double v); @Override void writeBytes(String str); @Override void writeChars(String s); @Override void writeUTF(String str); @Override void writeAsSerializedByteArray(Object v);  }
@Test public void streamerListRelease() throws IOException { final MsgStreamerList msgStreamer = (MsgStreamerList) createMsgStreamer(true); msgStreamer.writeMessage(); verify(pool, times(2)).releaseSenderBuffer(isA(ByteBuffer.class)); }
@Test public void streamerListReleaseWithException() throws IOException { final MsgStreamerList msgStreamer = (MsgStreamerList) createMsgStreamer(true); doThrow(new SSLException("")).when(connection1).sendPreserialized(any(ByteBuffer.class), any(Boolean.class), any(DistributionMessage.class)); msgStreamer.writeMessage(); verify(pool, times(2)).releaseSenderBuffer(isA(ByteBuffer.class)); }