lx/netapp
1
1
Fork 5
Code Issues 2 Pull requests 1 Releases 1 Wiki Activity

add streaming body to requests and responses #3

Merged
lx merged 64 commits from stream-body into main 2022-09-13 10:56:54 +00:00
Conversation 0 Commits 64 Files changed 22 +3 -8
1 changed files with 3 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit f0326607ee - Show all commits

11
examples/fullmesh.rs
View file

@ -142,6 +142,7 @@ impl Example {
example_field, example_field,
hex::encode(id) hex::encode(id)
); );
// Fake data stream with some delays in item production
let stream = let stream =
Box::pin(stream::iter([100, 200, 300, 400]).then(|x| async move { Box::pin(stream::iter([100, 200, 300, 400]).then(|x| async move {
tokio::time::sleep(Duration::from_millis(500)).await; tokio::time::sleep(Duration::from_millis(500)).await;
@ -191,16 +192,10 @@ impl StreamingEndpointHandler<ExampleMessage> for Example {
msg.msg() msg.msg()
); );
let source_stream = msg.take_stream().unwrap(); let source_stream = msg.take_stream().unwrap();
// Return same stream with 300ms delay
let new_stream = Box::pin(source_stream.then(|x| async move { let new_stream = Box::pin(source_stream.then(|x| async move {
info!(
"Handler: stream got bytes {:?}",
x.as_ref().map(|b| b.len())
);
tokio::time::sleep(Duration::from_millis(300)).await; tokio::time::sleep(Duration::from_millis(300)).await;
Ok(Bytes::from(vec![ x
10u8;
x.map(|b| b.len()).unwrap_or(1422) * 2
]))
})); }));
Resp::new(ExampleResponse { Resp::new(ExampleResponse {
example_field: false, example_field: false,