Upgrade to winnow 0.6

Author: snaggen

Cargo.toml

@@ -18,7 +18,7 @@ anyhow = "1.0"
 futures = "0.3"
 tokio = { version = "1", features = ["net"] }
 tokio-util = { version = "0.7", features = ["codec"] }
-winnow = "0.5"
+winnow = "0.6"
 
 [dev-dependencies]
 tokio = { version = "1", features = ["time", "macros", "rt-multi-thread"] }

src/client.rs

@@ -1,10 +1,12 @@
-use bytes::BytesMut;
+use bytes::{Buf, BytesMut};
 use futures::prelude::*;
 use futures::sink::SinkExt;
 
 use tokio::net::TcpStream;
 use tokio_util::codec::{Decoder, Encoder, Framed};
 use winnow::error::ErrMode;
+use winnow::stream::Offset;
+use winnow::Partial;
 
 pub type ClientTransport = Framed<TcpStream, ClientCodec>;
 
@@ -75,11 +77,14 @@ impl Decoder for ClientCodec {
     type Error = anyhow::Error;
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>> {
-        let item = match frame::parse_frame(&mut &src[..]) {
+        let buf = &mut Partial::new(src.chunk());
+        let item = match frame::parse_frame(buf) {
             Ok(frame) => Message::<FromServer>::from_frame(frame),
             Err(ErrMode::Incomplete(_)) => return Ok(None),
             Err(e) => bail!("Parse failed: {:?}", e),
         };
+        let len = buf.offset_from(&Partial::new(src.chunk()));
+        src.advance(len);
         item.map(Some)
     }
 }
@@ -96,3 +101,155 @@ impl Encoder<Message<ToServer>> for ClientCodec {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use futures::{future::ok, SinkExt, StreamExt, TryStreamExt};
+
+    use crate::{client, FromServer, ToServer};
+
+    // Test to send a message
+    #[tokio::test]
+    #[ignore]
+    async fn test_session() {
+        let mut conn = crate::client::connect(
+            "localhost:61613",
+            "/",
+            //None,None
+            Some("artemis".to_string()),
+            Some("artemis".to_string()),
+        )
+        .await
+        .expect("Default connection to localhost");
+        let msg = crate::Message {
+            content: ToServer::Send {
+                destination: "/test/a".to_string(),
+                transaction: None,
+                headers: Some(vec![("header-a".to_string(), "value-a".to_string())]),
+                body: Some("This is a test message".as_bytes().to_vec()),
+            },
+            extra_headers: vec![],
+        };
+        conn.send(msg).await.expect("Send a");
+        let msg = crate::Message {
+            content: ToServer::Send {
+                destination: "/test/b".to_string(),
+                transaction: None,
+                headers: Some(vec![("header-b".to_string(), "value-b".to_string())]),
+                body: Some("This is a another test message".as_bytes().to_vec()),
+            },
+            extra_headers: vec![],
+        };
+        conn.send(msg).await.expect("Send b");
+    }
+
+    // Test to recieve a message
+    #[tokio::test]
+    #[ignore]
+    async fn test_subscribe() {
+        let sub_msg = crate::client::subscribe("/test/a", "tjo");
+        let mut conn = crate::client::connect(
+            "localhost:61613",
+            "/",
+            //None,None
+            Some("artemis".to_string()),
+            Some("artemis".to_string()),
+        )
+        .await
+        .expect("Default connection to localhost");
+        conn.send(sub_msg).await.expect("Send subscribe");
+        let (_sink, stream) = conn.split();
+
+        let mut cnt = 0;
+        let _ = stream
+            .try_for_each(|item| {
+                println!("==== {cnt}");
+                cnt += 1;
+                if let FromServer::Message { body, .. } = item.content {
+                    println!(
+                        "Message received: {:?}",
+                        String::from_utf8_lossy(&body.unwrap())
+                    );
+                } else {
+                    println!("{:?}", item);
+                }
+                ok(())
+            })
+            .await;
+    }
+
+    // Test to send and recieve message
+    #[tokio::test]
+    #[ignore]
+    async fn test_send_subscribe() {
+        let conn = client::connect(
+            "127.0.0.1:61613",
+            "/".to_string(),
+            "artemis".to_string().into(),
+            "artemis".to_string().into(),
+        )
+        .await
+        .expect("Connect");
+
+        tokio::time::sleep(Duration::from_millis(200)).await;
+
+        let (mut sink, stream) = conn.split();
+
+        let fut1 = async move {
+            sink.send(client::subscribe("rusty", "myid")).await?;
+            println!("Subscribe sent");
+
+            tokio::time::sleep(Duration::from_millis(200)).await;
+
+            sink.send(
+                ToServer::Send {
+                    destination: "rusty".into(),
+                    transaction: None,
+                    headers: None,
+                    body: Some(b"Hello there rustaceans!".to_vec()),
+                }
+                .into(),
+            )
+            .await?;
+            println!("Message sent");
+
+            tokio::time::sleep(Duration::from_millis(200)).await;
+
+            sink.send(ToServer::Unsubscribe { id: "myid".into() }.into())
+                .await?;
+            println!("Unsubscribe sent");
+
+            tokio::time::sleep(Duration::from_millis(200)).await;
+
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            sink.send(ToServer::Disconnect { receipt: None }.into())
+                .await?;
+            println!("Disconnect sent");
+
+            Ok(())
+        };
+
+        // Listen from the main thread. Once the Disconnect message is sent from
+        // the sender thread, the server will disconnect the client and the future
+        // will resolve, ending the program
+        let fut2 = stream.try_for_each(|item| {
+            if let FromServer::Message { body, .. } = item.content {
+                println!(
+                    "Message received: {:?}",
+                    String::from_utf8_lossy(&body.unwrap())
+                );
+            } else {
+                //    println!("{:?}", item);
+            }
+            ok(())
+        });
+
+        futures::future::select(Box::pin(fut1), Box::pin(fut2))
+            .await
+            .factor_first()
+            .0
+            .expect("Select");
+    }
+}

src/frame.rs

@@ -5,7 +5,7 @@ use winnow::{
     ascii::{alpha1, line_ending, till_line_ending},
     combinator::{delimited, opt, repeat, separated_pair, terminated, trace},
     token::{literal, take, take_till, take_until},
-    PResult, Parser,
+    PResult, Parser, Partial,
 };
 
 use std::borrow::Cow;
@@ -101,7 +101,7 @@ fn is_empty_slice(s: &[u8]) -> Option<&[u8]> {
     }
 }
 
-pub(crate) fn parse_frame<'a>(input: &mut &'a [u8]) -> PResult<Frame<'a>> {
+pub fn parse_frame<'a>(input: &mut Partial<&'a [u8]>) -> PResult<Frame<'a>> {
     let (command, headers): (_, Vec<_>) = trace(
         "parse_frame",
         (
@@ -129,7 +129,7 @@ pub(crate) fn parse_frame<'a>(input: &mut &'a [u8]) -> PResult<Frame<'a>> {
     })
 }
 
-fn parse_header<'a>(input: &mut &'a [u8]) -> PResult<Header<'a>> {
+pub fn parse_header<'a>(input: &mut Partial<&'a [u8]>) -> PResult<Header<'a>> {
     trace(
         "parse_header",
         separated_pair(
@@ -546,7 +546,7 @@ login:user
 heart-beat:6,7
 passcode:password\n\n\x00"
             .to_vec();
-        let frame = parse_frame(&mut data.as_slice()).unwrap();
+        let frame = parse_frame(&mut Partial::new(data.as_slice())).unwrap();
         let headers_expect: Vec<(&[u8], &[u8])> = vec![
             (&b"accept-version"[..], &b"1.2"[..]),
             (b"host", b"datafeeds.here.co.uk"),
@@ -569,7 +569,7 @@ accept-version:1.2
 host:datafeeds.here.co.uk
 login:user
 passcode:password\n\n\x00";
-        let frame = parse_frame(&mut data.as_slice()).unwrap();
+        let frame = parse_frame(&mut Partial::new(data.as_slice())).unwrap();
         eprintln!("Frame: {frame:?}");
         let headers_expect: Vec<(&[u8], &[u8])> = vec![
             (&b"accept-version"[..], &b"1.2"[..]),
@@ -587,7 +587,7 @@ passcode:password\n\n\x00";
     /// https://stomp.github.io/stomp-specification-1.2.html#DISCONNECT
     fn parse_and_serialize_client_disconnect() {
         let data = b"DISCONNECT\nreceipt:77\n\n\x00";
-        let frame = parse_frame(&mut data.as_slice()).unwrap();
+        let frame = parse_frame(&mut Partial::new(data.as_slice())).unwrap();
         let headers_expect: Vec<(&[u8], &[u8])> = vec![(b"receipt", b"77")];
 
         assert_eq!(frame.command, b"DISCONNECT");
@@ -605,7 +605,7 @@ passcode:password\n\n\x00";
         let body = b"this body contains no nulls \n and \n newlines OK?";
         data.extend_from_slice(body);
         data.extend_from_slice(b"\x00");
-        let frame = parse_frame(&mut data.as_slice()).unwrap();
+        let frame = parse_frame(&mut Partial::new(data.as_slice())).unwrap();
         let headers_expect: Vec<(&[u8], &[u8])> = vec![(&b"destination"[..], &b"/queue/a"[..])];
         assert_eq!(frame.command, b"SEND");
         parse_and_serialize_to_server(&data, frame, headers_expect, Some(body));
@@ -623,7 +623,7 @@ passcode:password\n\n\x00";
         let body = "this body contains \x00 nulls \n and \r\n newlines \x00 OK?";
         let rest = format!("content-length:{}\n\n{}\x00", body.len(), body);
         data.extend_from_slice(rest.as_bytes());
-        let frame = parse_frame(&mut data.as_slice()).unwrap();
+        let frame = parse_frame(&mut Partial::new(data.as_slice())).unwrap();
         let headers_expect: Vec<(&[u8], &[u8])> = vec![
             (&b"destination"[..], &b"/queue/a"[..]),
             (b"content-type", b"text/html;charset=utf-8"),