How to communicate with gRPC server by stream, need some examples

[y0zong]

Hi, there is a storage service by gRPC(use tonic) and I want save the files uploaded by the client to it, blow is some fake code:

service Storage {
    rpc PostResource(stream RequestPayload) returns (stream ResponseCallback) {}
}
message ResourcePayload {
    string filename = 1;
    string type = 2;
}
message RequestPayload {
    oneof type {
        ResourcePayload file = 1;
        bytes data = 3;
    }
}

#[web::post("/file")]
pub async fn upload(
    grpc: State<Arc<Mutex<StorageRPC>>>,
    mut payload: Multipart,
) -> Result<impl Responder, Error> {
    let mut client = grpc.lock().await.connect().await;

    // 1. the stream will send to gPRC
    let stream: Stream

    while let Ok(Some(mut field)) = payload.try_next().await {
        // 2-1. get some info from field, for example `content-type` and filename
        while let Some(chunk) = field.next().await {
            // 2-2. combining file info and data to `stream`
        }
    }

    // 3. send stream payload to gRPC server
    let mut response = client
        .post_resource(Request::new(stream))
        .await
        .unwrap()
        .into_inner();

    while let Some(message) = response.message().await.unwrap() {
        // 4. throw some error if not success
        println!("info = {:?}", message);
    }

    Ok(HttpResponse::Ok())
}

there are many thread safety and ownership issues during the implementation and I have no idea about this
I would be very grateful if have some examples

[fafhrd91]

you cannot use ntex in multithreaded context. you need thread safe channel, send one side to tonic, and send chunks of data from ntex handler.

other option is to create tonic connection for each ntex worker.

[y0zong]

you are right, I take the thread channel way
Thank you!