Java 类javax.ws.rs.sse.SseEventSink 实例源码

@POST
@Path("domains/{id}")
@Produces(MediaType.SERVER_SENT_EVENTS)
public void startDomain(@PathParam("id") final String id, @Context SseEventSink eventSink) {

    executorService.submit(() -> {
        try {
            eventSink.send(sse.newEventBuilder().name("domain-progress").data(String.class, "starting domain " + id + " ...").build());
            Thread.sleep(200);
            eventSink.send(sse.newEvent("domain-progress", "50%"));
            Thread.sleep(200);
            eventSink.send(sse.newEvent("domain-progress", "60%"));
            Thread.sleep(200);
            eventSink.send(sse.newEvent("domain-progress", "70%"));
            Thread.sleep(200);
            eventSink.send(sse.newEvent("domain-progress", "99%"));
            Thread.sleep(200);
            eventSink.send(sse.newEvent("domain-progress", "Done."));
            eventSink.close();
        } catch (final InterruptedException e) {
            e.printStackTrace();
        }
    });
}

@POST
@Path("progress/{report_id}")
@Produces(MediaType.SERVER_SENT_EVENTS)
public void eventStream(@PathParam("report_id") final String id,
                        @Context SseEventSink es,
                        @Context Sse sse) {
    executorService.execute(() -> {
        try {
            eventSink.send(sse.newEventBuilder().name("report-progress")
                    .data(String.class, "Commencing process for report " + id + " ...").build());
            es.send(sse.newEvent("Progress", "25%"));
            Thread.sleep(500);
            es.send(sse.newEvent("Progress", "50%"));
            Thread.sleep(500);
            es.send(sse.newEvent("Progress", "75%"));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
}

@GET
@Path("subscribe")
@Produces(MediaType.SERVER_SENT_EVENTS)
public void subscribe(@Context SseEventSink eventSink) {
    eventSink.send(sse.newEvent("You are subscribed"));
    broadcaster.register(eventSink);
}

@GET
@Produces(MediaType.SERVER_SENT_EVENTS)
public void getMessageQueue(@Context SseEventSink eventSink) {
    synchronized (outputLock) {
        if (this.eventSink != null) {
            throw new IllegalStateException("Server sink already served.");
        }
        this.eventSink = eventSink;
    }
}

@GET
@Path("subscribe")
@Produces(MediaType.SERVER_SENT_EVENTS)
public void subscribe(@Context SseEventSink eventSink,
                      @Context Sse sse){
    eventSink.send(sse.newEvent("Welcome to the List!"));
    eventSink.send(sse.newEvent("Message One!"));
    eventSink.send(sse.newEvent("SERVER-NOTIFICATION", "Message Two!"));
    eventSink.send(sse.newEventBuilder()
                    .comment("Nice Test")
                    .name("SERVER-TEST")
                    .data("Some data...could be an object")
                    .build());
    eventSink.close();
}

@GET
@Lock(READ)
@Produces(MediaType.SERVER_SENT_EVENTS)
public void itemEvents(@HeaderParam(HttpHeaders.LAST_EVENT_ID_HEADER)
                       @DefaultValue("-1") int lastEventId,
                       @Context SseEventSink eventSink) {

    if (lastEventId >= 0)
        replayLastMessages(lastEventId, eventSink);

    sseBroadcaster.register(eventSink);
}

private void replayLastMessages(int lastEventId, SseEventSink eventSink) {
    try {
        for (int i = lastEventId; i < messages.size(); i++) {
            eventSink.send(createEvent(messages.get(i), i + 1));
        }
    } catch (Exception e) {
        throw new InternalServerErrorException("Could not replay messages ", e);
    }
}

public void deregister(String uuid) {
    LOG.log(Level.INFO, "deregister request:{0}", uuid);
    SseRequest req = requests.remove(uuid);
    try (SseEventSink eventSink = req.getEventSink()) {
        eventSink.close();
    }
}

@GET
@Produces(MediaType.SERVER_SENT_EVENTS)
public void eventStream(@Context Sse sse, @Context SseEventSink eventSink) {
    // Resource method is invoked when a client subscribes to an event stream.
    // That implies that sending events will most likely happen from different
    // context - thread / event handler / etc, so common implementation of the
    // resource method will store the eventSink instance and the application 
    // logic will retrieve it when an event should be emitted to the client.

    // sending events:
    eventSink.send(sse.newEvent("event1"));
}

@GET
//@Path("register")
@Produces(MediaType.SERVER_SENT_EVENTS)
public void register(@Context SseEventSink eventSink) {
    eventSink.send(sse.newEvent("welcome!"));
    sseBroadcaster.register(eventSink);
}

@GET
@Path("/{uuid}")
@Produces(SERVER_SENT_EVENTS)
public void doSseCall(@PathParam("uuid") String uuid, @Context SseEventSink sink, @Context Sse sse) {
    final OutboundSseEvent.Builder builder = sse.newEventBuilder();
    OutboundSseEvent event = builder.id(uuid)
            .data(SseModel.class, new SseModel("some model "+uuid))
            .build();
    sink.send(event);
    sink.close();
}

@Path("subscribe")
@GET
@Produces("text/event-stream")
public void subscribe(@Context SseEventSink eSink) {
    broadcaster.register(eSink);
}

public void register(SseEventSink eventSink){
    broadcaster.register(eventSink);
    System.out.println("Registered Event sink");
}

public SseRequest(Sse sse, SseEventSink eventSink) {
    this.sse = sse;
    this.eventSink = eventSink;
}

public SseEventSink getEventSink() {
    return eventSink;
}

public void setEventSink(SseEventSink eventSink) {
    this.eventSink = eventSink;
}

@GET
@Produces(MediaType.SERVER_SENT_EVENTS)
public void eventStreamCdi(@Context Sse sse, @Context SseEventSink eventSink) {
    handler.register(UUID.randomUUID().toString(), new SseRequest(sse, eventSink));
}

public SseEvent(SseEventSink sink, Sse sse, String id) {
    this.sink = sink;
    this.sse = sse;
    this.id = id;
}

@GET
@Path("{connectionId}")
@Produces(MediaType.SERVER_SENT_EVENTS)
public void onEvent(@Context SseEventSink sink, @PathParam("connectionId") final String id, @Context Sse sse) {
    event.fire(new SseEvent(sink, sse, id));
}