Akka의 EventStream에서 서버 측 이벤트를 어떻게 만들 수 있습니까?

Question

I 시도 : EventStream -> 소스 -> Akka HTTP (SSE)

내가 그것을보고,이 소스가 Akka의 HTTP complete(Source, ...) 에 의해 구체화되기 때문에, 작동하지 않을 수 있고, 구체화에 EventStream에서 메시지를 보낼 수 소스 나 (ActorRef 것을 얻을 수있는 방법이 있나요?)을 ActorRef 필요

---

내가 ActorPublisher 사용 GitHub의에서 해결책을 발견했습니다 : ActorPublisher 내부 API이기 때문에 https://github.com/calvinlfer/Akka-HTTP-Akka-Streams-Akka-Actors-Integration

을하지만, , 나는 여전히 깨끗한 해결책을 원하고있다.

Answer 1

당신은 다음과 같은 방식으로, ServerSentEvent 인스턴스에 이벤트 스트림 요소를 변환 Source 및 BroadcastHub.sink을 만들 Source.actorRef를 사용할 수 있습니다

val (sseActor, sseSource) = 
    Source.actorRef[EventStreamMessageOrWhatever](10, akka.stream.OverflowStrategy.dropTail) 
    .map(s => /* convert event stream elements to ServerSideEvent */) 
    .keepAlive(1.second,() => ServerSentEvent.heartbeat) 
    .toMat(BroadcastHub.sink[ServerSentEvent])(Keep.both) 
    .run() 

을 즉, 다운 스트림 수요, 메시지 (있는 경우, 이벤트 스트림 요소)가 구체화 된 ActorRef으로 전송되면 하류로 방출됩니다. 다운 스트림 요구가 없으면 메시지는 지정된 오버플로 전략으로 특정 수 (이 예에서는 버퍼 크기가 10)까지 버퍼링됩니다.

그런 다음 EventStream에 구체화 된 배우를 구독 할 수 있습니다 :

path("sse") { 
    get { 
    complete(sseSource) 
    } 
} 

참고이 방법에는 배압이 없다는 것을 :

eventStream.subscribe(sseActor, ...) 

그리고 구체화 된 Source 당신의 경로에 사용할 수 있습니다 .