XhrResourceBuilder.java 文件源码

private <T> void eventSourceSubscription(Subscriber<T> s) {
    final EventSource source = new EventSource(uri());
    final QueuedProducer<T> producer = new QueuedProducer<>(s);
    try {
        s.add(subscribeEventListener(source, "message", evt -> {
            producer.onNext(parse(Js.<MessageEvent<String>>cast(evt).data));
        }));
        s.add(subscribeEventListener(source, "open", evt -> {
            log.fine("Connection opened: " + uri());
        }));
        s.add(subscribeEventListener(source, "error", evt -> {
            log.log(Level.SEVERE, "Error: " + evt);
            if (source.readyState == source.CLOSED) {
                producer.onError(new RuntimeException("Event source error"));
            }
        }));
        s.setProducer(producer);
        s.add(Subscriptions.create(() -> {
            // hack because elemental API EventSource.close is missing
            Js.<MessagePort>uncheckedCast(source).close();
        }));
    } catch (Throwable e) {
        log.log(Level.FINE, "Received http error for: " + uri(), e);
        s.onError(new RuntimeException("Event source error", e));
    }
}