JerseySseBroadcaster extends JerseyPublisher<OutboundSseEvent> implements SseBroadcaster { @Override public void onClose(Consumer<SseEventSink> onClose) { if (onClose == null) { throw new IllegalArgumentException(LocalizationMessages.PARAM_NULL("onClose")); } onCloseListeners.add(onClose); } JerseySseBroadcaster();   JerseySseBroadcaster(final ExecutorService executorService); @Override void register(SseEventSink sseEventSink); @Override void onError(BiConsumer<SseEventSink, Throwable> onError); @Override void onClose(Consumer<SseEventSink> onClose); @Override CompletionStage<?> broadcast(final OutboundSseEvent event);  }
@Test public void testOnCloseNull() { try (JerseySseBroadcaster jerseySseBroadcaster = new JerseySseBroadcaster()) { thrown.expect(IllegalArgumentException.class); jerseySseBroadcaster.onClose(null); } }
@Test public void testOnClose() throws InterruptedException { try (JerseySseBroadcaster broadcaster = new JerseySseBroadcaster()) { final CountDownLatch latch = new CountDownLatch(1); final SseEventSink eventSink = new SseEventSink() { @Override public boolean isClosed() { return false; } @Override public CompletionStage<?> send(OutboundSseEvent event) { return null; } @Override public void close() { } }; broadcaster.register(eventSink); broadcaster.onClose((s) -> { if (s.equals(eventSink)) { latch.countDown(); } }); broadcaster.close(); Assert.assertTrue(latch.await(2000, TimeUnit.MILLISECONDS)); } }