SSE SseEmitter.completeWithError(e) 触发的处理逻辑

在 Java 客户端使用 OkHttp 监听 SSE（Server-Sent Events） 的情况下，当服务端调用 SseEmitter.completeWithError(e)，客户端会触发 EventSourceListener 的 onFailure() 方法（而不是 onError）。

---

1. 服务端（Spring Boot）调用 completeWithError

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@RestController
public class SseController {

    @GetMapping("/sse-stream")
    public SseEmitter streamEvents() {
        SseEmitter emitter = new SseEmitter();

        new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    emitter.send(SseEmitter.event().data("Event " + i));
                    Thread.sleep(1000);
                }
                // 模拟一个错误并主动关闭
                throw new RuntimeException("Server-side error!");
            } catch (Exception e) {
                emitter.completeWithError(e); // 触发客户端的 onFailure()
            }
        }).start();

        return emitter;
    }
}

---

2. 客户端（Java + OkHttp）监听 SSE

OkHttp 提供了 EventSource 类来监听 SSE 事件，并需要实现 EventSourceListener 来接收回调。

关键方法

• onOpen() – 连接建立时触发。
• onEvent() – 收到服务器事件时触发。
• onClosed() – 服务器主动关闭（emitter.complete()）时触发。
• onFailure() – 服务器调用 completeWithError() 或网络错误时触发。

代码示例

import okhttp3.*;
import okhttp3.sse.EventSource;
import okhttp3.sse.EventSourceListener;
import okhttp3.sse.EventSources;

public class SseClient {
    public static void main(String[] args) {
        OkHttpClient client = new OkHttpClient();
        Request request = new Request.Builder()
                .url("http://localhost:8080/sse-stream")
                .build();

        EventSource.Factory factory = EventSources.createFactory(client);
        factory.newEventSource(request, new EventSourceListener() {
            @Override
            public void onOpen(EventSource eventSource, Response response) {
                System.out.println("SSE connected!");
            }

            @Override
            public void onEvent(EventSource eventSource, String id, String type, String data) {
                System.out.println("Received event: " + data);
            }

            @Override
            public void onClosed(EventSource eventSource) {
                System.out.println("SSE closed by server.");
            }

            @Override
            public void onFailure(EventSource eventSource, Throwable t, Response response) {
                System.err.println("SSE error: " + t.getMessage());
                // 可以在这里重试
            }
        });
    }
}

---

3. 执行流程

1. 服务端：
   • 发送5条事件后，调用 emitter.completeWithError(e)。
2. 客户端：
   • 收到5条正常事件（onEvent()）。
   • 当服务端 completeWithError(e) 时，触发 onFailure()，并打印错误信息。

---

4. 总结

服务端行为	客户端（OkHttp）回调方法
emitter.send(data)	onEvent()
emitter.complete()	onClosed()
emitter.completeWithError(e)	onFailure()
网络断开/超时	onFailure()

因此，SseEmitter.completeWithError(e) 会触发客户端的 onFailure() 方法，而不是 onError（这是浏览器 EventSource 的行为）。