DeferredStreamMessage extends AbstractStreamMessage<T> { protected final void delegate(StreamMessage<T> upstream) { requireNonNull(upstream, "upstream"); if (!upstreamUpdater.compareAndSet(this, null, upstream)) { throw new IllegalStateException("upstream set already"); } final Throwable abortCause = this.abortCause; if (abortCause != null) { upstream.abort(abortCause); } if (!whenComplete().isDone()) { upstream.whenComplete().handle((unused, cause) -> { if (cause == null) { whenComplete().complete(null); } else { whenComplete().completeExceptionally(cause); } return null; }).exceptionally(CompletionActions::log); } safeOnSubscribeToUpstream(); }  final void close(); final void close(Throwable cause); @Override final boolean isOpen(); @Override final boolean isEmpty(); @Override final void abort(); @Override final void abort(Throwable cause);  }
@Test void testSetDelegate() { final DeferredStreamMessage<Object> m = new DeferredStreamMessage<>(); m.delegate(new DefaultStreamMessage<>()); assertThatThrownBy(() -> m.delegate(new DefaultStreamMessage<>())) .isInstanceOf(IllegalStateException.class); assertThatThrownBy(() -> m.delegate(null)).isInstanceOf(NullPointerException.class); }