title: Broadcasting with Jersey SSE: Detect closed connection
body: <p>I believe this question is not a duplicate of <a href="https://stackoverflow.com/questions/22275030/server-sent-event-with-jersey-eventoutput-is-not-closed-after-client-drops">Server sent event with Jersey: EventOutput is not closed after client drops</a>, but probably related to <a href="https://stackoverflow.com/questions/28349585/jersey-server-sent-events-write-to-broken-connection-does-not-throw-exception">Jersey Server-Sent Events - write to broken connection does not throw exception</a>.</p>
  
  <p>In <a href="https://jersey.java.net/documentation/latest/sse.html#d0e11704" rel="noreferrer">chapter 15.4.2</a> of the Jersey documentation, the <a href="https://jersey.java.net/apidocs/2.21/jersey/org/glassfish/jersey/media/sse/SseBroadcaster.html" rel="noreferrer">SseBroadcaster</a> is described:</p>
  
  <blockquote>
    <p>However, the SseBroadcaster internally identifies and handles also client disconnects. When a client closes the connection the broadcaster detects this and removes the stale connection from the internal collection of the registered EventOutputs as well as it frees all the server-side resources associated with the stale connection.</p>
  </blockquote>
  
  <p>I cannot confirm this. In the following testcase, I see the subclassed <code>SseBroadcaster</code>'s <code>onClose()</code> method never being called: not when the <code>EventInput</code> is closed, and not when another message is broadcasted.</p>
  
  <pre><code>public class NotificationsResourceTest extends JerseyTest {
      final static Logger log = LoggerFactory.getLogger(NotificationsResourceTest.class);
  
      final static CountingSseBroadcaster broadcaster = new CountingSseBroadcaster();
  
      public static class CountingSseBroadcaster extends SseBroadcaster { 
          final AtomicInteger connectionCounter = new AtomicInteger(0);
  
          public EventOutput createAndAttachEventOutput() {
              EventOutput output = new EventOutput();
              if (add(output)) {
                  int cons = connectionCounter.incrementAndGet();
                  log.debug("Active connection count: "+ cons);
              }
              return output;
          }
  
          @Override
          public void onClose(final ChunkedOutput&lt;OutboundEvent&gt; output) {
              int cons = connectionCounter.decrementAndGet();
              log.debug("A connection has been closed. Active connection count: "+ cons);
          }
  
          @Override
          public void onException(final ChunkedOutput&lt;OutboundEvent&gt; chunkedOutput, final Exception exception) {
              log.trace("An exception has been detected", exception);
          }
  
          public int getConnectionCount() {
              return connectionCounter.get();
          }
      }
  
      @Path("notifications")
      public static class NotificationsResource {
  
          @GET
          @Produces(SseFeature.SERVER_SENT_EVENTS)
          public EventOutput subscribe() {
              log.debug("New stream subscription");
  
              EventOutput eventOutput = broadcaster.createAndAttachEventOutput();
              return eventOutput;
          }
      }   
  
      @Override
      protected Application configure() {
          ResourceConfig config = new ResourceConfig(NotificationsResource.class);
          config.register(SseFeature.class);
  
          return config;
      }
  
  
      @Test
      public void test() throws Exception {
          // check that there are no connections
          assertEquals(0, broadcaster.getConnectionCount());
  
          // connect subscriber
          log.info("Connecting subscriber");
          EventInput eventInput = target("notifications").request().get(EventInput.class);
          assertFalse(eventInput.isClosed());
  
          // now there are connections
          assertEquals(1, broadcaster.getConnectionCount());
  
          // push data
          log.info("Broadcasting data");
          String payload = UUID.randomUUID().toString();
          OutboundEvent chunk = new OutboundEvent.Builder()
                  .mediaType(MediaType.TEXT_PLAIN_TYPE)
                  .name("message")
                  .data(payload)
                  .build();
          broadcaster.broadcast(chunk);
  
          // read data
          log.info("Reading data");
          InboundEvent inboundEvent = eventInput.read();
          assertNotNull(inboundEvent);
          assertEquals(payload, inboundEvent.readData());
  
          // close subscription 
          log.info("Closing subscription");
          eventInput.close();
          assertTrue(eventInput.isClosed());
  
          // at this point, the subscriber has disconnected itself, 
          // but jersey doesnt realise that
          assertEquals(1, broadcaster.getConnectionCount());
  
          // wait, give TCP a chance to close the connection
          log.debug("Sleeping for some time");
          Thread.sleep(10000);
  
          // push data again, this should really flush out the not-connected client
          log.info("Broadcasting data again");
          broadcaster.broadcast(chunk);
          Thread.sleep(100);
  
          // there is no subscriber anymore
          assertEquals(0, broadcaster.getConnectionCount());  // FAILS!
      }
  }
  </code></pre>
  
  <p>Maybe <code>JerseyTest</code> is not a good way to test this. In a less ... clinical setup, where a JavaScript <code>EventSource</code> is used, I see <code>onClose()</code> being called, but only after a message is broadcasted on the previously closed connection.</p>
  
  <p>What am I doing wrong?</p>
  
  <p>Why doesn't <code>SseBroadcaster</code> detect the closing of the connection by the client?</p>
  
  <p><strong>Follow-up</strong></p>
  
  <p>I've found <a href="https://java.net/jira/browse/JERSEY-2833" rel="noreferrer">JERSEY-2833</a> which was rejected with <em>Works as designed</em>:</p>
  
  <blockquote>
    <p>According to the Jersey Documentation in SSE chapter (<a href="https://jersey.java.net/documentation/latest/sse.html" rel="noreferrer">https://jersey.java.net/documentation/latest/sse.html</a>) in 15.4.1 it's mentioned that Jersey does not explicitly close the connection, it's the responsibility of the resource method or the client.</p>
  </blockquote>
  
  <p>What does that mean exactly? Should the resource enforce a timeout and kill all active and closed-by-client connections?</p>