### Question:
AbstractSubscription implements Subscription { public final void signal(boolean force) { if (force || demand > 0) { signal(); } } protected  AbstractSubscription(Subscriber<? super T> downstream, Executor executor); @Override final void request(long n); @Override final void cancel(); final void signal(boolean force); final void signalError(Throwable error);  }### Answer:
@Test void rejectedExecution() { Executor busy = r -> { throw new RejectedExecutionException(); }; var p = new SubmittableSubscription<>(new TestSubscriber<Integer>(), busy); assertThrows(RejectedExecutionException.class, () -> p.signal(true)); assertEquals(1, p.aborts); assertTrue(p.flowInterrupted); }
@Test void subscribeNoSignals() { var s = new TestSubscriber<Integer>(); var p = subscription(s); p.signal(true); s.awaitSubscribe(); assertNotNull(s.subscription); assertEquals(0, s.nexts); assertEquals(0, s.errors); assertEquals(0, s.completes); }
@Test void noItems() { var s = new TestSubscriber<Integer>(); var p = subscription(s); p.signal(true); p.complete(); s.awaitComplete(); assertEquals(0, s.nexts); assertEquals(0, s.errors); assertEquals(1, s.completes); }
@Test void errorOnSubscribe() { var s = new TestSubscriber<Integer>(); s.throwOnCall = true; var p = subscription(s); p.signal(true); s.awaitError(); assertEquals(0, s.nexts); assertEquals(1, s.errors); assertEquals(0, s.completes); }
@Test void oneItem() { var s = new TestSubscriber<Integer>(); var p = subscription(s); p.signal(true); p.items.offer(1); p.complete(); s.awaitComplete(); assertEquals(1, s.nexts); assertEquals(1, s.completes); }
@Test void errorOnNext() { var s = new TestSubscriber<Integer>(); var p = subscription(s); p.signal(true); s.awaitSubscribe(); p.submit(1); s.throwOnCall = true; p.submit(2); p.complete(); s.awaitComplete(); assertEquals(1, s.errors); assertEquals(0, s.completes); }
@Test void itemOrder() { var s = new TestSubscriber<Integer>(); var p = subscription(s); p.signal(true); for (int i = 1; i <= 20; ++i) p.submit(i); p.complete(); s.awaitComplete(); assertEquals(20, s.nexts); assertEquals(20, s.items.peekLast()); assertEquals(1, s.completes); }
@Test void requestOneReceiveOne() { var s = new TestSubscriber<Integer>(); var p = subscription(s); p.signal(true); s.awaitSubscribe(); s.request = 0L; p.submit(1); p.submit(2); p.complete(); s.awaitNext(1); assertEquals(1, s.nexts); }
@Test void abort_byErrorOnSubscribe() { var s = new TestSubscriber<Integer>(); s.throwOnCall = true; var p = subscription(s); p.signal(true); s.awaitError(); p.awaitAbort(); assertEquals(1, s.errors); assertEquals(1, p.aborts); assertTrue(p.flowInterrupted); }
@Test void abort_byErrorOnNext() { var s = new TestSubscriber<Integer>(); var p = subscription(s); p.signal(true); s.awaitSubscribe(); s.throwOnCall = true; p.submit(1); p.complete(); s.awaitError(); p.awaitAbort(); assertEquals(1, s.errors); assertEquals(1, p.aborts); assertTrue(p.flowInterrupted); }