### Question:
QueueToStreamUnicast implements Publisher<T> { @Override public void subscribe(final Subscriber<? super T> subscriber) { final SubscriptionImpl subscription = new SubscriptionImpl(); try { this.queue.startListener(new ReceiveQueueListener<T>() { long sendThisMany = 0; boolean stop = false; @Override public void empty() { initStreamState(); } private void initStreamState() { sendThisMany = subscription.count(sendThisMany); stop = subscription.stop.get(); } @Override public void limit() { initStreamState(); } @Override public void shutdown() { subscriber.onComplete(); } @Override public void idle() { initStreamState(); } @Override public void startBatch() { initStreamState(); } @Override public void init() { subscriber.onSubscribe(subscription); } @Override public void receive(T item) { waitForCountsIfNeeded(); if (stop) { queue.stop(); } subscriber.onNext(item); sendThisMany--; } private void waitForCountsIfNeeded() { if (sendThisMany == 0) { onSubscriptionEmpty.run(); for (int index = 0; index < 100_000; index++) { initStreamState(); if (sendThisMany > 0 || stop) break; } for (int index = 0; index < 10_000; index++) { initStreamState(); if (sendThisMany > 0 || stop) break; Thread.yield(); } for (int index = 0; index < 100; index++) { initStreamState(); if (sendThisMany > 0 || stop) break; Sys.sleep(1); } for (int index = 0; index < 100; index++) { initStreamState(); if (sendThisMany > 0 || stop) break; Sys.sleep(5); } while (sendThisMany == 0) { initStreamState(); if (sendThisMany > 0 || stop) break; Sys.sleep(15); } } } }); } catch (Exception ex) { subscriber.onError(ex); } } QueueToStreamUnicast(final Queue<T> queue);  QueueToStreamUnicast(final Queue<T> queue, final Runnable runnable); @Override void subscribe(final Subscriber<? super T> subscriber);  }### Answer:
@Test public void test() throws InterruptedException { final Queue<Trade> queue = QueueBuilder.queueBuilder().build(); final QueueToStreamUnicast<Trade> stream = new QueueToStreamUnicast<>(queue); final CopyOnWriteArrayList<Trade> trades = new CopyOnWriteArrayList<>(); final AtomicBoolean stop = new AtomicBoolean(); final AtomicReference<Throwable> error = new AtomicReference<>(); final AtomicReference<Subscription> subscription = new AtomicReference<>(); final CountDownLatch latch = new CountDownLatch(1); final CountDownLatch stopLatch = new CountDownLatch(1); stream.subscribe(new Subscriber<Trade>() { @Override public void onSubscribe(Subscription s) { s.request(10); subscription.set(s); } @Override public void onNext(Trade trade) { trades.add(trade); if (trades.size() == 10) { latch.countDown(); } } @Override public void onError(Throwable t) { error.set(t); } @Override public void onComplete() { stop.set(true); stopLatch.countDown(); } }); final SendQueue<Trade> tradeSendQueue = queue.sendQueue(); for (int index = 0; index < 20; index++) { tradeSendQueue.send(new Trade("TESLA", 100L + index)); } tradeSendQueue.flushSends(); Sys.sleep(100); latch.await(10, TimeUnit.SECONDS); assertEquals(10, trades.size()); assertEquals(false, stop.get()); assertNotNull(subscription.get()); subscription.get().cancel(); stopLatch.await(10, TimeUnit.SECONDS); assertEquals(true, stop.get()); }