How to use actix field stream by two consumers?

user_2516756 0
Viewed 337

I have an actix web service and would like to parse the contents of a multipart field while streaming with async-gcode and in addition store the contents e.g. in a database.

However, I have no clue how to feed in the stream to the Parser and at the same time collect the bytes into a Vec<u8> or a String.

The first problem I face is that field is a stream of actix::web::Bytes and not of u8.

#[post("/upload")]
pub async fn upload_job(
    mut payload: Multipart,
) -> Result<HttpResponse, Error> {
    let mut contents : Vec<u8> = Vec::new();
    while let Ok(Some(mut field)) = payload.try_next().await {
        let content_disp = field.content_disposition().unwrap();
        match content_disp.get_name().unwrap() {
            "file" => {
                while let Some(chunk) = field.next().await {
                    contents.append(&mut chunk.unwrap().to_vec());
                    // already parse the contents 
                    // and additionally store contents somewhere
                }
            }
            _ => (),
        }
    }
    Ok(HttpResponse::Ok().finish())
}

Any hint or suggestion is very much appreciated.

actix-web async-await rust
2 Answers

One of the options is to wrap field in a struct and implement Stream trait for it.

use actix_web::{HttpRequest, HttpResponse, Error};
use futures_util::stream::Stream;
use std::pin::Pin;
use actix_multipart::{Multipart, Field};
use futures::stream::{self, StreamExt};
use futures_util::TryStreamExt;
use std::task::{Context, Poll};
use async_gcode::{Parser, Error as PError};
use bytes::BytesMut;
use std::cell::RefCell;

pub struct Wrapper {
    field: Field,
    buffer: RefCell<BytesMut>,
    index: usize,
}

impl Wrapper {
    pub fn new(field: Field, buffer: RefCell<BytesMut>) -> Self {
        buffer.borrow_mut().truncate(0);
        Wrapper {
            field,
            buffer,
            index: 0
        }
    }
}

impl Stream for Wrapper {
    type Item = Result<u8, PError>;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<u8, PError>>> {
        if self.index == self.buffer.borrow().len() {
            match Pin::new(&mut self.field).poll_next(cx) {
                Poll::Ready(Some(Ok(chunk))) => self.buffer.get_mut().extend_from_slice(&chunk),
                Poll::Pending => return Poll::Pending,
                Poll::Ready(None) => return Poll::Ready(None),
                Poll::Ready(Some(Err(_))) => return Poll::Ready(Some(Err(PError::BadNumberFormat/* ??? */)))
            };
        } else {
            let b = self.buffer.borrow()[self.index];
            self.index += 1;
            return Poll::Ready(Some(Ok(b)));
        }
        Poll::Ready(None)
    }
}

#[post("/upload")]
pub async fn upload_job(
    mut payload: Multipart,
) -> Result<HttpResponse, Error> {
    while let Ok(Some(field)) = payload.try_next().await {
        let content_disp = field.content_disposition().unwrap();
        match content_disp.get_name().unwrap() {
            "file" => {
                let mut contents: RefCell<BytesMut> = RefCell::new(BytesMut::new());
                let mut w = Wrapper::new(field, contents.clone());
                let mut p = Parser::new(w);
                while let Some(res) = p.next().await {
                    // Do something with results
                };
                // Do something with the buffer
                let a = contents.get_mut()[0];
            }
            _ => (),
        }
    }
    Ok(HttpResponse::Ok().finish())
}

Copying the Bytes from the Field won't be necessary when Bytes::try_unsplit will be implemented. (https://github.com/tokio-rs/bytes/issues/287)

user_15415010

The answer from dmitryvm (thanks for your effort) showed me that there are actually two problems. At first, flatten the Bytes into u8's and, secondly, to "split" the stream into a buffer for later storage and the async-gcode parser.

This shows how I solved it:

#[post("/upload")]
pub async fn upload_job(
    mut payload: Multipart,
) -> Result<HttpResponse, Error> {
    let mut contents : Vec<u8> = Vec::new();
    while let Ok(Some(mut field)) = payload.try_next().await {
        let content_disp = field.content_disposition().unwrap();
        match content_disp.get_name().unwrap() {
            "file" => {
                let field_stream = field
                    .map_err(|_| async_gcode::Error::BadNumberFormat) // Translate error
                    .map_ok(|y| { // Translate Bytes into stream with Vec<u8>
                        contents.extend_from_slice(&y);   // Copy and store for later usage
                        stream::iter(y).map(Result::<_, async_gcode::Error>::Ok) 
                    })
                    .try_flatten(); // Flatten the streams of u8's
                let mut parser = Parser::new(field_stream);
                while let Some(gcode) = parser.next().await {
                    // Process result from parser
                }
            }
            _ => (),
        }
    }
    Ok(HttpResponse::Ok().finish())
}
user_25167560
Related