Bump tokio tungstenite to 0.26

Cargo.toml

@@ -24,7 +24,7 @@ bytes = { version = "1.7", features = ["serde"] }
 futures-core = "0.3"
 futures-util = { version = "0.3", default-features = false, features = ["std"] }
 tokio = "1.40"
-tokio-tungstenite = "0.24"
+tokio-tungstenite = "0.26"
 serde = { version = "1.0", features = ["derive"] }
 smallvec = { version = "1.13", features = ["union"] }
 serde_json = "1.0"

crates/engineioxide/src/packet.rs

@@ -110,12 +110,11 @@ impl Packet {
 }
 
 /// Serialize a [Packet] to a [String] according to the Engine.IO protocol
-impl TryInto<String> for Packet {
- type Error = Error;
- fn try_into(self) -> Result<String, Self::Error> {
- let len = self.get_size_hint(true);
+impl From<Packet> for String {
+ fn from(packet: Packet) -> String {
+ let len = packet.get_size_hint(true);
  let mut buffer = String::with_capacity(len);
- match self {
+ match packet {
  Packet::Open(open) => {
  buffer.push('0');
  buffer.push_str(&serde_json::to_string(&open).unwrap());
@@ -140,7 +139,12 @@ impl TryInto<String> for Packet {
  general_purpose::STANDARD.encode_string(data, &mut buffer);
  }
  };
- Ok(buffer)
+ buffer
+ }
+}
+impl From<Packet> for tokio_tungstenite::tungstenite::Utf8Bytes {
+ fn from(value: Packet) -> Self {
+ String::from(value).into()
  }
 }
 /// Deserialize a [Packet] from a [String] according to the Engine.IO protocol
@@ -176,6 +180,13 @@ impl TryFrom<Str> for Packet {
  Ok(res)
  }
 }
+impl TryFrom<tokio_tungstenite::tungstenite::Utf8Bytes> for Packet {
+ type Error = Error;
+ fn try_from(value: tokio_tungstenite::tungstenite::Utf8Bytes) -> Result<Self, Self::Error> {
+ // SAFETY: The utf8 bytes are guaranteed to be valid utf8
+ Packet::try_from(unsafe { Str::from_bytes_unchecked(value.into()) })
+ }
+}
 
 impl TryFrom<String> for Packet {
  type Error = Error;

crates/engineioxide/src/transport/polling/payload/encoder.rs

@@ -88,7 +88,7 @@ pub async fn v4_encoder(
  try_recv_packet(&mut rx, data.len() + PUNCTUATION_LEN, max_payload, true)
  {
  for packet in packets {
- let packet: String = packet.try_into()?;
+ let packet: String = packet.into();
 
  if !data.is_empty() {
  data.push(std::char::from_u32(PACKET_SEPARATOR_V4 as u32).unwrap());
@@ -101,7 +101,7 @@ pub async fn v4_encoder(
  if data.is_empty() {
  let packets = recv_packet(&mut rx).await?;
  for packet in packets {
- let packet: String = packet.try_into()?;
+ let packet: String = packet.into();
  data.push_str(&packet);
  }
  }
@@ -133,7 +133,7 @@ pub fn v3_bin_packet_encoder(packet: Packet, data: &mut bytes::BytesMut) -> Resu
  data.extend_from_slice(&bin); // raw data
  }
  packet => {
- let packet: String = packet.try_into()?;
+ let packet: String = packet.into();
  let len = itoa.format(packet.len());
  let len_len = len.len(); // len is guaranteed to be ascii
 
@@ -156,7 +156,7 @@ pub fn v3_bin_packet_encoder(packet: Packet, data: &mut bytes::BytesMut) -> Resu
 pub fn v3_string_packet_encoder(packet: Packet, data: &mut bytes::BytesMut) -> Result<(), Error> {
  use crate::transport::polling::payload::STRING_PACKET_SEPARATOR_V3;
  use bytes::BufMut;
- let packet: String = packet.try_into()?;
+ let packet: String = packet.into();
  let packet = format!(
  "{}{}{}",
  packet.chars().count(),

crates/engineioxide/src/transport/ws.rs

@@ -185,9 +185,9 @@ where
  Message::Binary(mut data) => {
  if socket.protocol == ProtocolVersion::V3 && !data.is_empty() {
  // The first byte is the message type, which we don't need.
-let _ = data.remove(0);
+data = data.split_off(1);
  }
- engine.handler.on_binary(data.into(), socket.clone());
+ engine.handler.on_binary(data, socket.clone());
  Ok(())
  }
  Message::Close(_) => break,
@@ -221,12 +221,16 @@ where
  ($item:ident) => {
  let res = match $item {
  Packet::Binary(bin) | Packet::BinaryV3(bin) => {
- let mut bin: Vec<u8> = bin.into();
  if socket.protocol == ProtocolVersion::V3 {
- // v3 protocol requires packet type as the first byte
- bin.insert(0, 0x04);
+ // v3 protocol requires packet type as the first byte.
+ // This requires a new buffer. This is OK as it is only for the V3 protocol.
+ let mut buff = Vec::with_capacity(bin.len() + 1);
+ buff.push(0x04);
+ buff.extend(bin);
+ tx.feed(Message::Binary(buff.into())).await
+ } else {
+ tx.feed(Message::Binary(bin)).await
  }
- tx.feed(Message::Binary(bin)).await
  }
  Packet::Close => {
  tx.send(Message::Close(None)).await.ok();
@@ -239,7 +243,7 @@ where
  Packet::Noop => Ok(()),
  _ => {
  let packet: String = $item.try_into().unwrap();
- tx.feed(Message::Text(packet)).await
+ tx.feed(Message::Text(packet.into())).await
  }
  };
  if let Err(_e) = res {
@@ -274,7 +278,7 @@ where
  S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
 {
  let packet = Packet::Open(OpenPacket::new(TransportType::Websocket, sid, config));
- ws.send(Message::Text(packet.try_into()?)).await?;
+ ws.send(Message::Text(packet.into())).await?;
  Ok(())
 }
 
@@ -320,8 +324,7 @@ where
  match Packet::try_from(msg)? {
  Packet::PingUpgrade => {
  // Respond with a PongUpgrade packet
- ws.send(Message::Text(Packet::PongUpgrade.try_into()?))
- .await?;
+ ws.send(Message::Text(Packet::PongUpgrade.into())).await?;
  }
  p => Err(Error::BadPacket(p))?,
  };

crates/socketioxide/tests/disconnect_reason.rs

@@ -237,9 +237,7 @@ pub async fn server_ns_close() {
  });
 
  let mut ws = create_ws_connection(12353).await;
- ws.send(Message::Text("40/test,{}".to_string()))
- .await
- .unwrap();
+ ws.send(Message::Text("40/test,{}".into())).await.unwrap();
  let data = tokio::time::timeout(Duration::from_millis(20), rx.recv())
  .await
  .expect("timeout waiting for DisconnectReason::ServerNSDisconnect")

crates/socketioxide/tests/fixture.rs

@@ -101,7 +101,7 @@ pub async fn create_ws_connection(port: u16) -> WebSocketStream<MaybeTlsStream<T
  .unwrap()
  .0;
 
- ws.send(Message::Text("40{}".to_string())).await.unwrap();
+ ws.send(Message::Text("40{}".into())).await.unwrap();
 
  ws
 }