AbstractStream implements Stream { protected final void onStreamAllocated() { checkState(listener() != null); synchronized (onReadyLock) { checkState(!allocated, "Already allocated"); allocated = true; } notifyIfReady(); } AbstractStream(WritableBufferAllocator bufferAllocator, int maxMessageSize); @VisibleForTesting  AbstractStream(MessageFramer framer, MessageDeframer deframer); @Nullable abstract IdT id(); int getOnReadyThreshold(); @Override void writeMessage(InputStream message); @Override final void setMessageCompression(boolean enable); @Override final void flush(); @Override boolean isReady(); void dispose(); @Override final void setCompressor(Compressor compressor); @Override final void setDecompressor(Decompressor decompressor); boolean canReceive(); boolean canSend(); @VisibleForTesting boolean isClosed(); @Override String toString(); static final int DEFAULT_ONREADY_THRESHOLD; }
@Test public void onStreamAllocated_shouldNotifyReady() { AbstractStream<Object> stream = new AbstractStreamBase<Object>(null); stream.onStreamAllocated(); verify(streamListener).onReady(); }