How to return multiple events/responses with actix-web

help
1

how can I write a jsonrpc subscription/request handler returning respose on events generated by notify watcher in actix web after single time registration. I will be receiving events in a loop and generate continuous response when a event arrive.
basically I want to do this. I dont know if it is possible or not.

use actix_web::{post, App, HttpResponse, HttpServer, Responder};

#[post("/")]
async fn handler(request: String, data: web::Data<AppState>) -> impl Responder {
    match request {
        //x => while let Ok(event) = data.rx.recv() {
        //    send event response
        //    upon next event again send event response in this function only
        //}
        //_ => return error
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| App::new().service(handler))
        .bind(("127.0.0.1", 7878))?
        .run()
        .await
}
2

It is not possible like that, as HTTP is request-response based.[1] You could upgrade the request to a websocket and send your events over the websocket channel instead. I'm not familiar with the internals of jsonrpc, but in the specification overview it is stated that jsonrpc is protocol agnostic, so it should be usable with websockets.

  1. See below. ↩ī¸Ž

1 Like
3

thank you very much

4

I agree with you websockets can be used in scenario. Further I was experimenting with Actix-web streaming. I want to know that this code below is idiomatic/correct/can be used?

I haven't tested fully yet I want to know if it is allowed otherwise I will choose Websockets.

use std::pin::Pin;
use std::task::{Context, Poll};

use actix_web::{get, web::Bytes, App, Error, HttpResponse, HttpServer};

pub struct TaskStream(u32);

impl futures::Stream for TaskStream {
    type Item = Result<Bytes, Error>;

    fn poll_next(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();
        this.0 += 1;
        // will recv events here and generate response
        Poll::Ready(Some(Ok(Bytes::copy_from_slice(
            &format!("response {}\n\n", this.0).into_bytes(),
        ))))
    }
}

#[get("/big-json")]
async fn big_json() -> HttpResponse {
    HttpResponse::Ok().streaming(TaskStream(0))
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| App::new().service(big_json))
        .bind("127.0.0.1:8080")?
        .run()
        .await
}

I m not good with words forgive any errors

1 Like
5

I've misread your question, my bad. Apologies for any confusion I've caused. I somehow thought you were trying to continuously receive some events from your request, generating a response for each. But you receive your events from somewhere on the server. Indeed that allows you to use HTTP streaming like you do with TaskStream.