Async FTP PUT

[mihaigalos]

Using async_ftp 6.0.0 and tokio 1.14.0, I am trying to asynchronously read a file and PUT it to a FTP server:

let mut ftp_stream = FtpStream::connect((*parsed_ftp).server.clone()).await.unwrap();
let mut file = tokio::fs::File::open(&input).await.unwrap();
let mut reader_stream = ReaderStream::new(file);

let mut async_stream = async_stream::stream! {
    while let Some(chunk) = reader_stream.next().await {
       yield chunk;
    }
};

How would it be best to read from the file stream and call put() on the chunk?
I thought of something like this, but this doesn't compile:

ftp_stream
    .put(&parsed_ftp.file, &mut async_stream)
    .await
    .unwrap();

Result:

&mut async_stream)
^^^^^^^^^^^^^^^^^ the trait `AsyncRead` is not implemented for `AsyncStream<_, impl futures_util::Future>`

[Darksonn]

Just use the tokio::fs::File object directly without the stream stuff.

[Darksonn]

This is kinda a hack, but you can put it in a tokio_util::StreamReader to convert back to AsyncRead.

[mihaigalos]

Reasonable, seems the Unpin is not implemented in that case:

let mut stream_reader = tokio_util::io::StreamReader::new(async_stream);
ftp_stream.put(&parsed_ftp.file, &mut stream_reader).await.unwrap();

Result:

.put(&parsed_ftp.file, &mut stream_reader)
^^^ within `tokio_util::io::stream_reader::_::__Origin<'_, AsyncStream<Result<bytes::bytes::Bytes, std::io::Error>, impl futures_util::Future>, bytes::bytes::Bytes>`, the trait `Unpin` is not implemented for `from_generator::GenFuture<[static generator@/home/mihai/.cargo/registry/src/github.com-1ecc6299db9ec823/async-stream-0.3.2/src/lib.rs:201:9: 201:67]>`

[Darksonn]

You can add a new line containing tokio::pin!(stream_reader); between those two lines.

[mihaigalos]

Thank you. That worked.
It's a bit scary to understand what is going on and why.
Do you have any recommendations on what to read to ramp up knowledge here? Books in particular.

[Darksonn]

There is some discussion of pinning in this chapter.