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

private static void consumeEventsViaSubscription() {
    try (final SseEventSource eventSource =
                 SseEventSource.target(target)
                         .build()) {

        eventSource.register(System.out::println);
        eventSource.open();

        for (int counter = 0; counter < 5; counter++) {
            target.request().post(Entity.text("message " + counter));
        }

        Thread.sleep(1000); // make sure all the events have time to arrive
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

public final static void main(String[] args) {

        WebTarget target = ClientBuilder.newClient().target("http://localhost:8080/jaxrs-sse/rest/events");

        try (SseEventSource eventSource = SseEventSource.target(target).build()) {

            // EventSource#register(Consumer<InboundSseEvent>)
            // Registered event handler will print the received message.
            eventSource.register(System.out::println);

            // Subscribe to the event stream.
            eventSource.open();

            // Consume events for just 500 ms
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

@Test
public void testCloseConnection() throws InterruptedException {
    //given
    ServerSentEventService ServerSentEventService = Application.getInstance(ServerSentEventService.class);
    Config config = Application.getInstance(Config.class);
    Client client = ClientBuilder.newClient();

    //when
    WebTarget webTarget = client.target("http://" + config.getConnectorHttpHost() + ":" + config.getConnectorHttpPort() + "/sse");
    SseEventSource sseEventSource = SseEventSource.target(webTarget).build();
    sseEventSource.register((sseEvent) -> {eventData = sseEvent.readData();}, (e) -> e.printStackTrace());
    sseEventSource.open();
    ServerSentEventService.close("/sse");
    sseEventSource.close();
    client.close();

    //then
    assertThat(ServerSentEventService.getConnections("/sse"), not(nullValue()));
    assertThat(ServerSentEventService.getConnections("/sse").size(), equalTo(0));
}

@Test
public void testSendData() throws InterruptedException {
    //given
    Config config = Application.getInstance(Config.class);
    Client client = ClientBuilder.newClient();
    String data = UUID.randomUUID().toString();

    //when
    WebTarget webTarget = client.target("http://" + config.getConnectorHttpHost() + ":" + config.getConnectorHttpPort() + "/sse");
    SseEventSource sseEventSource = SseEventSource.target(webTarget).build();
    sseEventSource.register((sseEvent) -> {eventData = sseEvent.readData();}, (e) -> e.printStackTrace());
    sseEventSource.open();

       //then
       Application.getInstance(ServerSentEventService.class).send("/sse", data);
       await().atMost(2,  TimeUnit.SECONDS).untilAsserted(() -> assertThat(eventData, equalTo(data)));
       sseEventSource.close();
       client.close();
}

@Test
public void shouldBeAbleToRetrieveRestEndpoint() throws Exception {
    WebTarget target = ClientBuilder.newClient().register(JohnzonExtension.class).target(uri+"/sse/{uuid}").resolveTemplate("uuid", UUID.randomUUID().toString());
    List<SseModel> receivedModels = new ArrayList<>();
    try (SseEventSource eventSource = SseEventSource.target(target).build()) {
        eventSource.register(event -> {
            SseModel body = event.readData(SseModel.class, MediaType.APPLICATION_JSON_TYPE);
            System.out.println("Received "+body.getName());
            receivedModels.add(body);
        }, System.out::println);
        eventSource.open();
        // Give the SSE stream some time to collect all events
        Thread.sleep(1000);
    }
    assertFalse(receivedModels.isEmpty());
}

@Test
public void testFireSseEventsAsync() throws InterruptedException {
    final WebTarget target = createWebTarget("/sse/" + UUID.randomUUID());
    final Collection<String> messages = new ArrayList<>();

    try (SseEventSource eventSource = SseEventSource.target(target).build()) {
        eventSource.register(new Consumer<InboundSseEvent>() {
            @Override
            public void accept(InboundSseEvent e) {
                System.out.println("New event...");
                messages.add(e.readData());
            }
        }, System.out::println);
        eventSource.open();
    }

    // wait for messages to come in
    Thread.sleep(8000);

    messages.forEach(System.out::println);
    assertThat(messages).hasSize(4);
}

public void ClientAPIExample() {

        WebTarget target = ClientBuilder.newClient().target("http://localhost:8080/jax-rs-2-1/sse/");
        try (SseEventSource source = SseEventSource.target(target).build()) {
            source.register(System.out::println);
            source.open();
        }

    }

public void connect(Consumer<String> dataConsumer) {
    eventSource = SseEventSource.target(target).build();

    eventSource.register(
            item -> dataConsumer.accept(item.readData()),
            Throwable::printStackTrace,
            () -> System.out.println("completed"));

    eventSource.open();
}