Stream to SSE from another endpoint (upload) #446

AlexMikhalev · 2022-12-07T21:39:36Z

AlexMikhalev
Dec 7, 2022

Hello all, firstly - poem is great, exactly what I need to move from python fast API.
The use case I am struggling to implement with poem open API:
Upload file - stream status of background worker after uploading into SSE, in more details:

file upload (using open API/upload example, works),
upload kick-offs bash script (tokio child, works), which runs for over 100 seconds,
I would like to stream status updates from the stdout buffer into SSE.
I tried to follow https://blog.logrocket.com/how-to-build-a-websocket-server-with-rust/ and mix it with open API. They create a channel sender inside the struct, which later wraps around arc mutex. I tried to add pub sender: Option<mpsc::UnboundedSender<std::result::Result<String, poem::Error>>>, into users_crud for checking, and it fails with the trait ToJSON is not implemented for UnboundedSender<Result<std::string::String, poem::Error>> . mpsc is tokio mpsc.

So, what's the right way to trigger the streaming of events from one open API handler to another?

Answered by sunli829

Dec 10, 2022

Execute bash script with POST

curl -X 'POST' 'http://localhost:3000/api/exec' -H 'Content-Type: text/plain; charset=utf-8' -d 'ls' -v

Get the outputs with SSE:

curl 'http://localhost:3000/api/output/afdec250-b5a6-42fe-bd37-d14aeadf9cf4' -H 'Content-Type: text/plain; charset=utf-8' -v

Cargo.toml

[package]
name = "ws_example"
version = "0.1.0"
edition = "2021"

[dependencies]
poem = { version = "1.3.49", features = ["session"] }
poem-openapi = { version = "2.0.21", features = ["swagger-ui", "uuid"] }
tokio = { version = "1.21.2", features = [
    "rt-multi-thread",
    "macros",
    "net",
    "sync",
    "process",
    "io-util",
] }
tokio-stream = { version = "0.1.11", features = ["sync"]…

View full answer

sunli829 · 2022-12-08T00:21:15Z

sunli829
Dec 8, 2022
Maintainer

I wrote an example, it's easy:

try it with

curl -X 'POST' 'http://localhost:3000/api/exec' -H 'Content-Type: text/plain; charset=utf-8' -d 'ls' -v

Cargo.toml

[package]
name = "exec"
version = "0.1.0"
edition = "2021"

[dependencies]
poem = { version = "1.3.49", features = ["session"] }
poem-openapi = { version = "2.0.21", features = ["swagger-ui"] }
tokio = { version = "1.21.2", features = [
    "rt-multi-thread",
    "macros",
    "net",
    "sync",
    "process",
    "io-util",
] }
tokio-stream = { version = "0.1.11", features = ["sync"] }

main.rs

use std::process::Stdio;

use poem::{error::BadRequest, listener::TcpListener, Result, Route, Server};
use poem_openapi::{
    payload::{EventStream, PlainText},
    OpenApi, OpenApiService,
};
use tokio::{
    io::{AsyncBufReadExt, BufReader},
    process::Command,
    sync::mpsc,
};
use tokio_stream::wrappers::UnboundedReceiverStream;

struct Api;

#[OpenApi]
impl Api {
    #[oai(path = "/exec", method = "post")]
    async fn exec(
        &self,
        command: PlainText<String>,
    ) -> Result<EventStream<UnboundedReceiverStream<String>>> {
        let mut child = Command::new(command.0)
            .stdout(Stdio::piped())
            .spawn()
            .map_err(BadRequest)?;
        let (tx, rx) = mpsc::unbounded_channel();

        tokio::spawn(async move {
            if let Some(stdout) = child.stdout.take() {
                let mut reader = BufReader::new(stdout);
                let mut line = String::new();

                loop {
                    if matches!(reader.read_line(&mut line).await, Ok(0) | Err(_)) {
                        break;
                    }
                    _ = tx.send(line.clone());
                }
            }
        });

        Ok(EventStream::new(UnboundedReceiverStream::new(rx)))
    }
}

#[tokio::main]
async fn main() -> Result<(), std::io::Error> {
    let app = OpenApiService::new(Api, "demo", "1.0").server("http://localhost:3000/api");
    let ui = app.swagger_ui();

    Server::new(TcpListener::bind("127.0.0.1:3000"))
        .name("hello-world")
        .run(Route::new().nest("/api", app).nest("/", ui))
        .await
}

5 replies

AlexMikhalev Dec 8, 2022
Author

That's a great reply. Thank you. Highly appreciate your help and work on Poem.

AlexMikhalev Dec 8, 2022
Author

I checked it worked and then went through Mozilla documentation: SSE doesn't support post endpoint, and the above example illustrates the issue: post into one endpoint, but stream into SSE shall be read from the get endpoint.

sunli829 Dec 9, 2022
Maintainer

I checked it worked and then went through Mozilla documentation: SSE doesn't support post endpoint, and the above example illustrates the issue: post into one endpoint, but stream into SSE shall be read from the get endpoint.

you are right!

AlexMikhalev Dec 9, 2022
Author

My attempt to solve it was to add the sender to users_crud example:

/// Create user schema
#[derive(Debug, Object, Clone)]
struct User {
    /// Id
    #[oai(read_only)]
    id: i64,
    /// Name
    #[oai(validator(max_length = 64))]
    name: String,
    /// Password
    #[oai(validator(max_length = 32))]
    password: Password,
    email: Email,
    pub sender: Option<mpsc::UnboundedSender<std::result::Result<String, poem::Error>>>,
    
}

it fails with the trait ToJSON is not implemented for UnboundedSender<Result<std::string::String, poem::Error>>.
Shall we use AddData or is anything better?

AlexMikhalev Dec 9, 2022
Author

To reiterate - I have two endpoints, one for POST /upload returns JSON ok and another /events/:user_id returns SSE, the question and problem is how to pass a stream of data between two endpoints.

sunli829 · 2022-12-10T03:21:34Z

sunli829
Dec 10, 2022
Maintainer

Execute bash script with POST

curl -X 'POST' 'http://localhost:3000/api/exec' -H 'Content-Type: text/plain; charset=utf-8' -d 'ls' -v

Get the outputs with SSE:

curl 'http://localhost:3000/api/output/afdec250-b5a6-42fe-bd37-d14aeadf9cf4' -H 'Content-Type: text/plain; charset=utf-8' -v

Cargo.toml

[package]
name = "ws_example"
version = "0.1.0"
edition = "2021"

[dependencies]
poem = { version = "1.3.49", features = ["session"] }
poem-openapi = { version = "2.0.21", features = ["swagger-ui", "uuid"] }
tokio = { version = "1.21.2", features = [
    "rt-multi-thread",
    "macros",
    "net",
    "sync",
    "process",
    "io-util",
] }
tokio-stream = { version = "0.1.11", features = ["sync"] }
uuid = { version = "1.2.2", features = ["v4"] }

main.rs

use std::{collections::HashMap, process::Stdio};

use poem::{
    error::BadRequest, http::StatusCode, listener::TcpListener, Error, Result, Route, Server,
};
use poem_openapi::{
    param::Path,
    payload::{EventStream, Json, PlainText},
    OpenApi, OpenApiService,
};
use tokio::{
    io::{AsyncBufReadExt, BufReader},
    process::Command,
    sync::{mpsc, Mutex},
};
use tokio_stream::wrappers::UnboundedReceiverStream;
use uuid::Uuid;

type OutputStream = EventStream<UnboundedReceiverStream<String>>;

#[derive(Default)]
struct Api {
    outputs: Mutex<HashMap<Uuid, OutputStream>>,
}

#[OpenApi]
impl Api {
    #[oai(path = "/exec", method = "post")]
    async fn exec(&self, command: PlainText<String>) -> Result<Json<Uuid>> {
        let mut child = Command::new(command.0)
            .stdout(Stdio::piped())
            .spawn()
            .map_err(BadRequest)?;
        let (tx, rx) = mpsc::unbounded_channel();

        tokio::spawn(async move {
            if let Some(stdout) = child.stdout.take() {
                let mut reader = BufReader::new(stdout);
                let mut line = String::new();

                loop {
                    if matches!(reader.read_line(&mut line).await, Ok(0) | Err(_)) {
                        break;
                    }
                    _ = tx.send(line.clone());
                }
            }
        });

        let id = Uuid::new_v4();
        self.outputs
            .lock()
            .await
            .insert(id, EventStream::new(UnboundedReceiverStream::new(rx)));
        Ok(Json(id))
    }

    #[oai(path = "/output/:id", method = "get")]
    async fn output(&self, id: Path<Uuid>) -> Result<OutputStream> {
        self.outputs
            .lock()
            .await
            .remove(&id)
            .ok_or_else(|| Error::from_status(StatusCode::NOT_FOUND))
    }
}

#[tokio::main]
async fn main() -> Result<(), std::io::Error> {
    let app =
        OpenApiService::new(Api::default(), "demo", "1.0").server("http://localhost:3000/api");
    let ui = app.swagger_ui();

    Server::new(TcpListener::bind("127.0.0.1:3000"))
        .name("hello-world")
        .run(Route::new().nest("/api", app).nest("/", ui))
        .await
}

1 reply

AlexMikhalev Dec 10, 2022
Author

Great example. That's exactly what I need. Thank you for your help and work on poem.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Stream to SSE from another endpoint (upload) #446

{{title}}

Replies: 2 comments 6 replies

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

{{title}}

Select a reply

Stream to SSE from another endpoint (upload) #446

AlexMikhalev Dec 7, 2022

Execute bash script with POST

Get the outputs with SSE:

Cargo.toml

Replies: 2 comments · 6 replies

sunli829 Dec 8, 2022 Maintainer

try it with

Cargo.toml

main.rs

AlexMikhalev Dec 8, 2022 Author

AlexMikhalev Dec 8, 2022 Author

sunli829 Dec 9, 2022 Maintainer

AlexMikhalev Dec 9, 2022 Author

AlexMikhalev Dec 9, 2022 Author

sunli829 Dec 10, 2022 Maintainer

Execute bash script with POST

Get the outputs with SSE:

Cargo.toml

main.rs

AlexMikhalev Dec 10, 2022 Author

AlexMikhalev
Dec 7, 2022

Replies: 2 comments 6 replies

sunli829
Dec 8, 2022
Maintainer

AlexMikhalev Dec 8, 2022
Author

AlexMikhalev Dec 8, 2022
Author

sunli829 Dec 9, 2022
Maintainer

AlexMikhalev Dec 9, 2022
Author

AlexMikhalev Dec 9, 2022
Author

sunli829
Dec 10, 2022
Maintainer

AlexMikhalev Dec 10, 2022
Author