How to test server sent events with Spring?

Question

I've implemented a controller with a method returning an SseEmitter and now I want to test it. The only way I could find so far is the following:

@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = {SsePaymentReceivedController.class, AutomatBackendContextInitializer.class, EventBusImpl.class})
@WebAppConfiguration
public class SsePaymentReceivedControllerIntegrationTest {

@Inject
WebApplicationContext context;
@Inject
SsePaymentReceivedController sseCoinController;
@Inject
EventBusImpl eventBus;

MockMvc mockMvc;

@Before
public void setUpMockMvc() {
    this.mockMvc = MockMvcBuilders.webAppContextSetup(this.context).build();
}

@Test
public void subscriptionToSseChannelIsFine() throws Exception {
    MvcResult result = mockMvc.perform(get("/sse/payment"))
            .andExpect(status().isOk())
            .andReturn();
    eventBus.fireNotification(new PaymentReceivedNotification("50", Currency.EURO));
    LinkedHashSet<ResponseBodyEmitter.DataWithMediaType> emitters =
            (LinkedHashSet<ResponseBodyEmitter.DataWithMediaType>)Whitebox.getInternalState(((SseEmitter)result.getModelAndView().getModel().get("sseEmitter")), "earlySendAttempts");
    final Iterator<ResponseBodyEmitter.DataWithMediaType> iterator = emitters.iterator();

    ResponseBodyEmitter.DataWithMediaType dataField = iterator.next();
    assertEquals("data:", dataField.getData());

    ResponseBodyEmitter.DataWithMediaType valueField = iterator.next();
    assertEquals("{\"remainingAmount\":\"50\",\"currency\":\"EURO\"}", valueField.getData());

    ResponseBodyEmitter.DataWithMediaType lastField = iterator.next();
    assertEquals("\n\n", lastField.getData());
}
}

There must be an approach better than inspecting the internals of the returned model and I'm looking for that - any ideas?

Answer 1

Spring 5.0 provides WebTestClient and ProjectReactor provides step verifier for testing the Server Sent Event (SSE).

Need to create WebTestClient object and make a call to endpoint. Below is an example:

//Example in Kotlin
        val testResult = webTestClient
                .get()
                .uri(URI.create("/api/sse"))
                .accept(MediaType.TEXT_EVENT_STREAM)
                .exchange()
                .expectStatus().isOk
                .expectHeader().contentTypeCompatibleWith(MediaType.TEXT_EVENT_STREAM)
                .returnResult(ServerSentEvent::class.java)
                .responseBody  //It results FluxExchangeResult<ServerSentEvent<*>>

        StepVerifier
                .create(testResult.map { it.id() })
                .expectSubscription()
                .expectNext("1", "2", "3")
                .verifyComplete()

StepVerifier provides rich set of methods which can be used for different situations for event streaming with Flux publisher.

Note: ServerSentEvent is model provided by SpringWebFlux framework for implementing SSE. It has fields as per SSE specification like, ID, data etc.

Answer 2

The best option I found is to get the content of the response :

String str1 = result.getResponse().getContentAsString()
// remove data: and \n\n
String str2 = str1.substring(5, str1.length() - 2);
new JsonPathExpectationsHelper("$.remainingAmount").assertValue(str2, "50");
new JsonPathExpectationsHelper("$.currency").assertValue(str2, "EURO");

Answer 3

You can use the WebTestClient on Spring Framework 5.0 or later.

Answer 4

A good idea might be to use an SSE client. You can use JAX WS RS Client with Jersey EventSource to achieve that.

With javax.ws.rs-api 2.0.1 and jersey-media-sse 2.25.1, here's how to open an SSE connection:

Client client = ClientBuilder
        .newBuilder()
        .register(SseFeature.class)
        .build();

EventSource eventSource = EventSource
        .target(client.target("http://localhost:8080/feed/123456"))
        .reconnectingEvery(1, TimeUnit.SECONDS)
        .build();

EventListener listener = inboundEvent -> {

    log.info("Event received: id -> {}, name -> {}, data -> {}", inboundEvent.getId(),
             inboundEvent.getName(), inboundEvent.readData(String.class));

    // Process events ...
};
eventSource.register(listener);
eventSource.open();

So if you want to create tools for testing, you can write a little helper class like this one:

package org.demo.service.sse;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;

import org.glassfish.jersey.media.sse.EventListener;
import org.glassfish.jersey.media.sse.EventSource;
import org.glassfish.jersey.media.sse.InboundEvent;
import org.glassfish.jersey.media.sse.SseFeature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SseTestClient {

    private static final Logger log = LoggerFactory.getLogger(SseTestClient.class);

    private final String targetUrl;
    private final List<InboundEvent> receivedEvents = new ArrayList<>();

    private String clientName = "SSE client";
    private EventSource eventSource;
    private int currentEventPos = 0;

    public SseTestClient( String targetUrl ) {

        this.targetUrl = targetUrl;
    }

    public SseTestClient setClientName( String clientName ) {

        this.clientName = clientName;
        return this;
    }

    public SseTestClient connectFeed() {

        log.info("[{}] Initializing EventSource...", clientName);

        // Create SSE client and server

        Client client = ClientBuilder
                .newBuilder()
                .register(SseFeature.class)
                .build();

        eventSource = EventSource
                .target(client.target(targetUrl))
                .reconnectingEvery(300, TimeUnit.SECONDS) // Large number so that any disconnection will break tests
                .build();

        EventListener listener = inboundEvent -> {

            log.info("[{}] Event received: name -> {}, data -> {}",
                     clientName, inboundEvent.getName(), inboundEvent.readData(String.class));

            receivedEvents.add(inboundEvent);
        };
        eventSource.register(listener);
        eventSource.open();

        log.info("[{}] EventSource connection opened", clientName);

        return this;
    }

    public int getTotalEventCount() {

        return receivedEvents.size();
    }

    public int getNewEventCount() {

        return receivedEvents.size() - currentEventPos;
    }

    public boolean hasNewEvent() {

        return currentEventPos < receivedEvents.size();
    }

    public InboundEvent getNextEvent() {

        if (currentEventPos >= receivedEvents.size()) {
            return null;
        }
        InboundEvent currentEvent = receivedEvents.get(currentEventPos);
        ++currentEventPos;
        return currentEvent;
    }

    public SseTestClient closeFeed() {

        if (eventSource != null) {
            eventSource.close();
        }
        return this;
    }
}

Which makes writing tests much easier:

SseTestClient sseClient = new SseTestClient("http://localhost:8080/feed/123456").connectFeed();

Thread.sleep(1000);

assertTrue(sseClient.hasNewEvent());
assertEquals(2, sseClient.getNewEventCount());
// ...

Hope it helps!