集成测试 Spring SseEmitters

Question

我一直在寻找有关如何最好地测试返回 SseEmitters 的 Spring MVC 控制器方法的提示。我的想法很短，但是有一个试错解决方案，可以针对异步线程行为进行测试。下面是示例代码，只是为了演示概念，可能有一两个错字：

控制器类：

@Autowired
Publisher<MyResponse> responsePublisher;

@RequestMapping("/mypath")
public SseEmitter index() throws IOException {
    SseEmitter emitter = new SseEmitter();
    Observable<MyResponse> responseObservable = RxReactiveStreams.toObservable(responsePublisher);

    responseObservable.subscribe(
            response -> {
                try {
                    emitter.send(response);
               } catch (IOException ex) {
                    emitter.completeWithError(ex);
               }
            },
            error -> {
                emitter.completeWithError(error);
            },
            emitter::complete
    );

    return emitter;
}

测试类：

//A threaded dummy publisher to demonstrate async properties.
//Sends 2 responses with a 250ms pause in between.
protected static class MockPublisher implements Publisher<MyResponse> {
    @Override
    public void subscribe(Subscriber<? super MyResponse> subscriber) {
        new Thread() {
            @Override
            public void run() {
                try {
                    subscriber.onNext(response1);
                    Thread.sleep(250);
                    subscriber.onNext(response2);
                } catch (InterruptedException ex) {
                }
                subscriber.onComplete();
            }
        }.start();
    }
}

//Assume @Configuration that autowires the above mock publisher in the controller.

//Tests the output of the controller method.
@Test
public void testSseEmitter() throws Exception {
    String path = "http://localhost/mypath/";
    String expectedContent = "data:" + response1.toString() + "\n\n" +
                             "data:" + response2.toString() + "\n\n");

    //Trial-and-Error attempts at testing this SseEmitter mechanism have yielded the following:
    //- Returning an SseEmitter triggers 'asyncStarted'
    //- Calling 'asyncResult' forces the test to wait for the process to complete
    //- However, there is no actual 'asyncResult' to test.  Instead, the content is checked for the published data.
    mockMvc.perform(get(path).contentType(MediaType.ALL))
        .andExpect(status().isOk())
        .andExpect(request().asyncStarted())
        .andExpect(request().asyncResult(nullValue()))
        .andExpect(header().string("Content-Type", "text/event-stream"))
        .andExpect(content().string(expectedContent))
}

如 cmets 中所述，调用 asyncResult() 以确保发布者完成其工作并在测试完成之前发送两个响应。没有它，内容检查会因为内容中只有一个响应而失败。但是没有实际结果要检查，因此 asyncResult 为空。

我的具体问题是，是否有更好、更精确的方法来强制测试等待异步进程完成，而不是这里等待不存在的 asyncResult 的 klugie 方法。我更广泛的问题是，与这些异步函数相比，是否有其他库或 Spring 方法更适合于此。谢谢！

Answer 1

这是一个更通用的答案，因为它旨在测试将永远运行的 SseEmitter，但会在给定超时后断开与 SSE 流的连接。

至于与 MVC 不同的方法，正如 @ErinDrummond 对 OP 所评论的那样，您可能需要研究 WebFlux。

这是一个最小的例子。可能想要扩展请求的标头、不同的匹配器，或者单独处理流输出。

它正在设置一个延迟线程来断开与 SSE 流的连接，这将允许执行断言。

@Autowired
MockMvc mockMvc;


@Test
public void testSseEmitter(){

    ScheduledExecutorService execService = Executors.newScheduledThreadPool(1);
    String streamUri = "/your-get-uri");
    long timeout = 500L;
    TimeUnit timeUnit = TimeUnit.MILLISECONDS;

    MvcResult result = mockMvc.perform(get(streamURI)
            .andExpect(request().asyncStarted()).andReturn();

    MockAsyncContext asyncContext = (MockAsyncContext) result.getRequest().getAsyncContext();
    execService.schedule(() -> {
        for (AsyncListener listener : asyncContext.getListeners())
            try {
                listener.onTimeout(null);
            } catch (IOException e) {
                e.printStackTrace();
            }
    }, timeout, timeUnit);

    result.getAsyncResult();

    // assertions, e.g. response body as string contains "xyz"
    mvc.perform(asyncDispatch(result)).andExpect(content().string(containsString("xyz")));
}