Canlı Veri — Polling, SSE ve WebSocket

1
@RestController
2
class TransferStreamController {
3

4
    private final TransferFeed feed;
5

6
    TransferStreamController(TransferFeed feed) {
7
        this.feed = feed;
8
    }
9

10
    @GetMapping(path = "/accounts/{id}/transfers/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
11
    SseEmitter stream(@PathVariable String id,
12
                      @RequestHeader(name = "Last-Event-ID", required = false) String lastSeen) {
13
        SseEmitter emitter = new SseEmitter(Duration.ofMinutes(30).toMillis());
14
        feed.missedSince(id, lastSeen).forEach(transfer -> feed.send(emitter, transfer));   // fill the gap first
15
        feed.subscribe(id, emitter);
16
        return emitter;
17
    }
18
}

1
@Component
2
class TransferFeed {
3

4
    private final Map<String, Set<SseEmitter>> listeners = new ConcurrentHashMap<>();
5
    private final TransferRepository transfers;
6

7
    TransferFeed(TransferRepository transfers) {
8
        this.transfers = transfers;
9
    }
10

11
    void subscribe(String accountId, SseEmitter emitter) {
12
        Set<SseEmitter> set = listeners.computeIfAbsent(accountId, key -> ConcurrentHashMap.newKeySet());
13
        set.add(emitter);
14
        emitter.onCompletion(() -> set.remove(emitter));
15
        emitter.onTimeout(() -> set.remove(emitter));
16
        emitter.onError(error -> set.remove(emitter));
17
    }
18

19
    List<IncomingTransfer> missedSince(String accountId, String lastSeen) {
20
        return lastSeen == null ? List.of() : transfers.findAfter(accountId, Long.parseLong(lastSeen));
21
    }
22

23
    @EventListener
24
    void on(IncomingTransfer transfer) {
25
        listeners.getOrDefault(transfer.accountId(), Set.of()).forEach(emitter -> send(emitter, transfer));
26
    }
27

28
    void forEachEmitter(Consumer<SseEmitter> action) {
29
        listeners.values().forEach(set -> set.forEach(action));
30
    }
31

32
    void send(SseEmitter emitter, IncomingTransfer transfer) {
33
        try {
34
            emitter.send(SseEmitter.event().id(String.valueOf(transfer.sequence())).name("transfer").data(transfer));
35
        } catch (IOException gone) {
36
            emitter.completeWithError(gone);   // the browser left; it will reconnect if it wants to
37
        }
38
    }
39
}

1
@Component
2
class Heartbeat {
3

4
    private final TransferFeed feed;
5

6
    Heartbeat(TransferFeed feed) {
7
        this.feed = feed;
8
    }
9

10
    /** A comment line every 15 seconds keeps proxies from closing a quiet stream. */
11
    @Scheduled(fixedRate = 15_000)
12
    void beat() {
13
        feed.forEachEmitter(emitter -> {
14
            try {
15
                emitter.send(SseEmitter.event().comment("ping"));
16
            } catch (IOException gone) {
17
                emitter.completeWithError(gone);
18
            }
19
        });
20
    }
21
}

1
export function watchTransfers(accountId: string, onTransfer: (t: IncomingTransfer) => void): () => void {
2
  // EventSource reconnects on its own and sends Last-Event-ID for us.
3
  const source = new EventSource(`/accounts/${accountId}/transfers/stream`, { withCredentials: true });
4
  source.addEventListener('transfer', (event) => onTransfer(JSON.parse(event.data) as IncomingTransfer));
5
  return () => source.close();
6
}