官术网_书友最值得收藏!

Exposing the SSE endpoint

The next step requires adding the TemperatureController class with the @RestController annotation, which means that the component is used for HTTP communication, as shown in the following code:

@RestController
public class TemperatureController {
private final Set<SseEmitter> clients = // (1)
new CopyOnWriteArraySet<>();

@RequestMapping(
value = "/temperature-stream", // (2)
method = RequestMethod.GET)
public SseEmitter events(HttpServletRequest request) { // (3)
SseEmitter emitter = new SseEmitter(); // (4)
clients.add(emitter); // (5)

// Remove emitter from clients on error or disconnect
emitter.onTimeout(() -> clients.remove(emitter)); // (6)
emitter.onCompletion(() -> clients.remove(emitter)); // (7)

return emitter; // (8)
}
@Async // (9)
@EventListener // (10)
public void handleMessage(Temperature temperature) { // (11)
List<SseEmitter> deadEmitters = new ArrayList<>(); // (12)
clients.forEach(emitter -> {
try {
emitter.send(temperature, MediaType.APPLICATION_JSON); // (13)
} catch (Exception ignore) {
deadEmitters.add(emitter); // (14)
}
});
clients.removeAll(deadEmitters); // (15)
}
}

Now, to understand the logic of the TemperatureController class, we need to describe the SseEmitter. Spring Web MVC provides that class with the sole purpose of sending SSE events. When a request-handling method returns the SseEmitter instance, the actual request processing continues until SseEnitter.complete(), an error, or a timeout occurs.

The TemperatureController provides one request handler (3) for the URI /temperature-stream (2) and returns the SseEmitter (8). In the case when a client requests that URI, we create and return the new SseEmitter instance (4) with its previous registration in the list of the active clients (5). Furthermore, the SseEmitter constructor may consume the timeout parameter.

For the clients' collection, we may use the CopyOnWriteArraySet class from the java.util.concurrent package (1). Such an implementation allows us to modify the list and iterate over it at the same time. When a web client opens a new SSE session, we add a new emitter to the clients' collection. The SseEmitter removes itself from the clients' list when it has finished processing or has reached timeout (6) (7).

Now, having a communication channel with clients means that we need to be able to receive events about temperature changes. For that purpose, our class has a handleMessage() method (11). It is decorated with the @EventListener annotation (10) in order to receive events from Spring. This framework will invoke the handleMessage() method only when receiving Temperature events, as this type of method's argument is known as temperature. The @Async annotation (9) marks a method as a candidate for the asynchronous execution, so it is invoked in the manually configured thread pool. The handleMessage() method receives a new temperature event and asynchronously sends it to all clients in JSON format in parallel for each event (13). Also, when sending to individual emitters, we track all failing ones (14) and remove them from the list of the active clients (15). Such an approach makes it possible to spot clients that are not operational anymore. Unfortunately, SseEmitter does not provide any callback for handling errors, and can be done by handling errors thrown by the send() method only.

主站蜘蛛池模板: 巴彦县| 邮箱| 莒南县| 夹江县| 开平市| 绍兴市| 呼图壁县| 德昌县| 五指山市| 镇江市| 雷波县| 桐庐县| 海原县| 唐河县| 广平县| 顺昌县| 海淀区| 闽清县| 白河县| 秀山| 深州市| 陈巴尔虎旗| 潞城市| 朝阳市| 亳州市| 罗源县| 铁力市| 滨州市| 建德市| 辽宁省| 万全县| 金寨县| 壶关县| 黑山县| 鹤岗市| 晋城| 罗城| 乌什县| 武胜县| 潜山县| 共和县|