我认为这个问题不是Server sent event with Jersey: EventOutput is not closed after client drops的重复,但可能与Jersey Server-Sent Events - write to broken connection does not throw exception相关。
也许
我做错了什么?
为什么
跟进:
我找到了JERSEY-2833,它因“按设计工作”而被拒绝:
根据SSE章节中的Jersey文档(https://jersey.java.net/documentation/latest/sse.html)中15.4.1的说明,Jersey不会明确地关闭连接,这是资源方法或客户端的责任。
那是什么意思?资源是否应强制执行超时并结束所有活动和由客户端关闭的连接?
在Jersey文档的第15.4.2章节中,描述了SseBroadcaster:
然而,SseBroadcaster内部也识别并处理客户端断开连接。当客户端关闭连接时,广播器检测到这一情况,并将陈旧的连接从已注册的EventOutputs的内部集合中删除,同时释放与陈旧连接相关联的所有服务器端资源。
我无法确认这一点。在以下测试用例中,我看到子类化的SseBroadcaster
的onClose()
方法从未被调用:不会在EventInput
关闭时,也不会在广播另一条消息时。
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<OutboundEvent> output) {
int cons = connectionCounter.decrementAndGet();
log.debug("A connection has been closed. Active connection count: "+ cons);
}
@Override
public void onException(final ChunkedOutput<OutboundEvent> 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!
}
}
也许
JerseyTest
不是测试这个的好方法。在一个不那么严格的环境下,使用JavaScript EventSource
时,我看到onClose()
被调用,但只有在先前关闭连接后广播消息之后才会发生。我做错了什么?
为什么
SseBroadcaster
无法检测到客户端关闭连接?跟进:
我找到了JERSEY-2833,它因“按设计工作”而被拒绝:
根据SSE章节中的Jersey文档(https://jersey.java.net/documentation/latest/sse.html)中15.4.1的说明,Jersey不会明确地关闭连接,这是资源方法或客户端的责任。
那是什么意思?资源是否应强制执行超时并结束所有活动和由客户端关闭的连接?