diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/message.rs | 11 |
1 files changed, 4 insertions, 7 deletions
diff --git a/src/message.rs b/src/message.rs index 56e6e8e..629992d 100644 --- a/src/message.rs +++ b/src/message.rs @@ -313,14 +313,13 @@ impl ReqEnc { buf.put(&self.telemetry_id[..]); buf.put_u32(self.msg.len() as u32); - buf.put(&self.msg[..]); let header = buf.freeze(); if let Some(stream) = self.stream { - Box::pin(futures::stream::once(async move { Ok(header) }).chain(stream)) + Box::pin(futures::stream::iter([Ok(header), Ok(self.msg)]).chain(stream)) } else { - Box::pin(futures::stream::once(async move { Ok(header) })) + Box::pin(futures::stream::iter([Ok(header), Ok(self.msg)])) } } @@ -387,16 +386,14 @@ impl RespEnc { let mut buf = BytesMut::with_capacity(msg.len() + 8); buf.put_u8(0); - buf.put_u32(msg.len() as u32); - buf.put(&msg[..]); let header = buf.freeze(); if let Some(stream) = stream { - Box::pin(futures::stream::once(async move { Ok(header) }).chain(stream)) + Box::pin(futures::stream::iter([Ok(header), Ok(msg)]).chain(stream)) } else { - Box::pin(futures::stream::once(async move { Ok(header) })) + Box::pin(futures::stream::iter([Ok(header), Ok(msg)])) } } RespEnc::Error { code, message } => { |