Come inviare un messaggio in un flusso reattivo da un sottoscrittore a un editore in una connessione socket Web

Question

La mia applicazione ha un'interfaccia Akka-Websocket. La presa web è composta da un attore-sottoscrittore e un editore attore. L'utente gestisce i comandi, inviandoli all'attore corrispondente. L'editore ascolta il flusso di eventi e pubblica le informazioni di aggiornamento sullo stream (e quindi sul client). Funziona bene

La mia domanda: come è possibile che il Sottoscrittore invii un evento allo stream? Ad esempio per confermare l'esecuzione di un comando ricevuto.

public class WebSocketApp extends HttpApp { 

    private static final Gson gson = new Gson(); 

    @Override 
    public Route createRoute() { 
    return get(
     path("metrics").route(handleWebSocketMessages(metrics())) 
     ); 
    } 

    private Flow<Message, Message, ?> metrics() { 
    Sink<Message, ActorRef> metricsSink = Sink.actorSubscriber(WebSocketCommandSubscriber.props()); 
    Source<Message, ActorRef> metricsSource = 
     Source.actorPublisher(WebSocketDataPublisherActor.props()) 
     .map((measurementData) -> TextMessage.create(gson.toJson(measurementData))); 
    return Flow.fromSinkAndSource(metricsSink, metricsSource); 
    } 
} 

Una bella soluzione potrebbe essere, che la sottoscrizione attore (il WebSocketCommandSubscriber attore del codice di cui sopra) potrebbe inviare un messaggio al flusso come sender().tell(...) ...

Answer 1

No, non è possibile, non direttamente, almeno. Gli stream sono sempre unidirezionali: tutti i messaggi fluiscono in una direzione, mentre la domanda per loro scorre nella direzione opposta. È necessario passare i messaggi di conferma dal sink al source affinché quest'ultimo possa restituirlo al client, ad esempio registrando l'attore di origine nell'attore sink. Potrebbe assomigliare a questo:

Flow.fromSinkAndSourceMat(metricsSink, metricsSource, (sinkActor, sourceActor) -> { 
    sinkActor.tell(new RegisterSource(sourceActor), null); 
}) 

Poi, dopo il tuo attore lavandino riceve RegisterSource messaggio, è possibile inviare i messaggi di conferma alla prevista ActorRef, che provvederanno poi a inoltrarle al flusso di output.