Enable tungstenite by default

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "easee"
-version = "0.1.0"
+version = "0.1.1"
 edition = "2021"
 authors = ["Maxime Augier <max@xolus.net>"]
 description = "Rust bindings for the Easee cloud API for EV charging devices"
@@ -15,13 +15,13 @@ categories = ["api-bindings"]
 
 [dependencies]
 chrono = { version = "0.4.38", features = ["serde"] }
-futures-util = { version = "0.3.30", features = ["futures-sink"] }
-reqwest = { version = "0.12.7", features = ["json"] }
 serde = { version = "1.0.204", features = ["derive"] }
 serde_json = "1.0.121"
 serde_repr = "0.1.19"
 thiserror = "1.0.63"
-tokio = "1.39.3"
-tokio-tungstenite = { version = "0.23.1", features = ["tokio-rustls", "rustls-tls-native-roots"] }
 tracing = "0.1.40"
 tungstenite = { version = "0.23.0", optional = true, features = ["rustls-tls-native-roots"] }
+ureq = { version = "2.10.0", features = ["json"] }
+
+[features]
+default = ["tungstenite"]

README.md

@@ -6,18 +6,24 @@ Work in progress.
 
  - Authn/z
    - [x] Authentication and token retrieval
-   - [x] Persistence of tokens
+   - [ ] Persistence of tokens
  - Core functionality
    - [x] Enumerate sites and chargers
    - [x] Read energy meter
    - [x] Read charger status
-   - [x] Control charging (start/pause/resume/stop)
-   - [x] Control dynamic current limits
+   - [ ] Control charging (start/pause/resume/stop)
+   - [ ] Control dynamic current limits
  - Event stream
    - [x] Websocket connection (raw SignalR messages)
-   - [x] Event decoding
+   - [ ] Event decoding
  - Ergonomics
    - [ ] Enums for protocol constants
    - [ ] Proper SignalR support with Tokio
 
+ 
+ 
+
+
+
+
 [1]: https://easee.com

src/api.rs

@@ -1,5 +1,7 @@
 use std::{
-    io, ops::{Add, Mul, Sub}, time::{Duration, Instant, SystemTime, UNIX_EPOCH}
+    io,
+    ops::{Add, Mul, Sub},
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
 };
 
 use serde::{de::DeserializeOwned, Deserialize, Deserializer, Serialize};
@@ -7,14 +9,11 @@ use serde_repr::Deserialize_repr;
 use thiserror::Error;
 use tracing::{debug, info, instrument};
 
-pub use reqwest::{self, StatusCode};
-
 pub struct Context {
     auth_header: String,
     refresh_token: String,
     token_expiration: Instant,
     on_refresh: Option<Box<dyn FnMut(&mut Self) + Send>>,
-    client: reqwest::Client,
 }
 
 impl std::fmt::Debug for Context {
@@ -317,7 +316,7 @@ pub enum ApiError {
 
     /// HTTP call failed (404, etc)
     #[error("ureq")]
-    HTTP(#[source] Box<reqwest::Error>),
+    Ureq(#[source] Box<ureq::Error>),
 
     /// HTTP call succeeded but the returned JSON document didn't match the expected format
     #[error("unexpected data: {1} when processing {0}")]
@@ -335,20 +334,20 @@ pub enum ApiError {
     InvalidID(String),
 }
 
-impl From<reqwest::Error> for ApiError {
-    fn from(value: reqwest::Error) -> Self {
-        ApiError::HTTP(Box::new(value))
+impl From<ureq::Error> for ApiError {
+    fn from(value: ureq::Error) -> Self {
+        ApiError::Ureq(Box::new(value))
     }
 }
 
 trait JsonExplicitError {
     /// Explicitely report the received JSON object we failed to parse
-    async fn into_json_with_error<T: DeserializeOwned>(self) -> Result<T, ApiError>;
+    fn into_json_with_error<T: DeserializeOwned>(self) -> Result<T, ApiError>;
 }
 
-impl JsonExplicitError for reqwest::Response {
-    async fn into_json_with_error<T: DeserializeOwned>(self) -> Result<T, ApiError> {
-        let resp: serde_json::Value = self.json().await?;
+impl JsonExplicitError for ureq::Response {
+    fn into_json_with_error<T: DeserializeOwned>(self) -> Result<T, ApiError> {
+        let resp: serde_json::Value = self.into_json()?;
         let parsed = T::deserialize(&resp);
         parsed.map_err(|e| ApiError::UnexpectedData(resp, e))
     }
@@ -364,13 +363,12 @@ pub enum TokenParseError {
 }
 
 impl Context {
-    fn from_login_response(resp: LoginResponse, client: reqwest::Client) -> Self {
+    fn from_login_response(resp: LoginResponse) -> Self {
         Self {
             auth_header: format!("Bearer {}", &resp.access_token),
             refresh_token: resp.refresh_token,
             token_expiration: (Instant::now() + Duration::from_secs(resp.expires_in as u64)),
             on_refresh: None,
-            client,
         }
     }
 
@@ -391,7 +389,6 @@ impl Context {
             refresh_token: refresh.to_owned(),
             token_expiration,
             on_refresh: None,
-            client: reqwest::Client::new(),
         })
     }
 
@@ -413,7 +410,7 @@ impl Context {
     }
 
     /// Retrieve access tokens online, by logging in with the provided credentials
-    pub async fn from_login(user: &str, password: &str) -> Result<Self, ApiError> {
+    pub fn from_login(user: &str, password: &str) -> Result<Self, ApiError> {
         #[derive(Serialize)]
         #[serde(rename_all = "camelCase")]
         struct Params<'t> {
@@ -421,26 +418,23 @@ impl Context {
             password: &'t str,
         }
 
-        let client = reqwest::Client::new();
-
         info!("Logging into API");
         let url: String = format!("{}accounts/login", API_BASE);
-        let resp: LoginResponse = client.post(&url)
-            .json(&Params {
+        let resp: LoginResponse = ureq::post(&url)
+            .send_json(Params {
                 user_name: user,
                 password,
-            })
-            .send().await?
-            .into_json_with_error().await?;
+            })?
+            .into_json_with_error()?;
 
-        Ok(Self::from_login_response(resp, client))
+        Ok(Self::from_login_response(resp))
     }
 
     /// Check if the token has reached its expiration date
-    async fn check_expired(&mut self) -> Result<(), ApiError> {
+    fn check_expired(&mut self) -> Result<(), ApiError> {
         if self.token_expiration < Instant::now() {
             debug!("Token has expired");
-            self.refresh_token().await?;
+            self.refresh_token()?;
         }
         Ok(())
     }
@@ -450,7 +444,7 @@ impl Context {
     }
 
     /// Use the refresh token to refresh credentials
-    pub async fn refresh_token(&mut self) -> Result<(), ApiError> {
+    pub fn refresh_token(&mut self) -> Result<(), ApiError> {
         #[derive(Serialize)]
         #[serde(rename_all = "camelCase")]
         struct Params<'t> {
@@ -462,52 +456,51 @@ impl Context {
             refresh_token: &self.refresh_token,
         };
         let url = format!("{}accounts/refresh_token", API_BASE);
-        let resp: LoginResponse = self.client.post(&url)
-            .header("Content-type", "application/json")
-            .json(&params)
-            .send().await?
-            .into_json_with_error().await?;
+        let resp: LoginResponse = ureq::post(&url)
+            .set("Content-type", "application/json")
+            .send_json(params)?
+            .into_json_with_error()?;
 
-        *self = Self::from_login_response(resp, self.client.clone());
+        *self = Self::from_login_response(resp);
         Ok(())
     }
 
     /// List all sites available to the user
-    pub async fn sites(&mut self) -> Result<Vec<Site>, ApiError> {
-        self.get("sites").await
+    pub fn sites(&mut self) -> Result<Vec<Site>, ApiError> {
+        self.get("sites")
     }
 
-    pub async fn site(&mut self, id: i32) -> Result<SiteDetails, ApiError> {
-        self.get(&format!("sites/{id}")).await
+    pub fn site(&mut self, id: i32) -> Result<SiteDetails, ApiError> {
+        self.get(&format!("sites/{id}"))
     }
 
     /// List all chargers available to the user
-    pub async fn chargers(&mut self) -> Result<Vec<Charger>, ApiError> {
-        self.get("chargers").await
+    pub fn chargers(&mut self) -> Result<Vec<Charger>, ApiError> {
+        self.get("chargers")
     }
 
-    pub async fn charger(&mut self, id: &str) -> Result<Charger, ApiError> {
+    pub fn charger(&mut self, id: &str) -> Result<Charger, ApiError> {
         if !id.chars().all(char::is_alphanumeric) {
             return Err(ApiError::InvalidID(id.to_owned()));
         }
-        self.get(&format!("chargers/{}", id)).await
+        self.get(&format!("chargers/{}", id))
     }
 
-    pub async fn circuit(&mut self, site_id: u32, circuit_id: u32) -> Result<Circuit, ApiError> {
-        self.get(&format!("site/{site_id}/circuit/{circuit_id}")).await
+    pub fn circuit(&mut self, site_id: u32, circuit_id: u32) -> Result<Circuit, ApiError> {
+        self.get(&format!("site/{site_id}/circuit/{circuit_id}"))
     }
 
-    pub async fn circuit_dynamic_current(
+    pub fn circuit_dynamic_current(
         &mut self,
         site_id: u32,
         circuit_id: u32,
     ) -> Result<Triphase, ApiError> {
         self.get(&format!(
             "sites/{site_id}/circuits/{circuit_id}/dynamicCurrent"
-        )).await
+        ))
     }
 
-    pub async fn set_circuit_dynamic_current(
+    pub fn set_circuit_dynamic_current(
         &mut self,
         site_id: u32,
         circuit_id: u32,
@@ -516,67 +509,65 @@ impl Context {
         self.post(
             &format!("sites/{site_id}/circuits/{circuit_id}/dynamicCurrent"),
             &current,
-        ).await
+        )
     }
 
     #[instrument]
-    async fn get<T: DeserializeOwned>(&mut self, path: &str) -> Result<T, ApiError> {
-        self.check_expired().await?;
+    fn get<T: DeserializeOwned>(&mut self, path: &str) -> Result<T, ApiError> {
+        self.check_expired()?;
         let url: String = format!("{}{}", API_BASE, path);
+        let req = ureq::get(&url)
+            .set("Accept", "application/json")
+            .set("Authorization", &self.auth_header);
 
-        let req = self.client.get(url)
-            .header("Accept", "application/json")
-            .header("Authorization", &self.auth_header)
-            .build()?;
-
-        let mut resp = self.client.execute(req.try_clone().unwrap()).await?;
+        let mut resp = req.clone().call()?;
 
         if resp.status() == 401 {
-            self.refresh_token().await?;
-            resp = self.client.execute(req).await?
+            self.refresh_token()?;
+            resp = req.call()?
         }
 
-        resp.into_json_with_error().await
+        resp.into_json_with_error()
     }
 
-    async fn maybe_get<T: DeserializeOwned>(&mut self, path: &str) -> Result<Option<T>, ApiError> {
-        match self.get(path).await {
+    fn maybe_get<T: DeserializeOwned>(&mut self, path: &str) -> Result<Option<T>, ApiError> {
+        match self.get(path) {
             Ok(r) => Ok(Some(r)),
-            Err(ApiError::HTTP(e)) if e.status() == Some(StatusCode::NOT_FOUND)=> Ok(None),
+            Err(ApiError::Ureq(e)) => match &*e {
+                ureq::Error::Status(404, _) => Ok(None),
+                _ => Err(ApiError::Ureq(e)),
+            },
             Err(other) => Err(other),
         }
     }
 
-    pub(crate) async fn post<T: DeserializeOwned, P: Serialize>(
+    pub(crate) fn post<T: DeserializeOwned, P: Serialize>(
         &mut self,
         path: &str,
         params: &P,
     ) -> Result<T, ApiError> {
         let url: String = format!("{}{}", API_BASE, path);
-        self.post_raw(&url, params).await
+        self.post_raw(&url, params)
     }
 
-    pub(crate) async fn post_raw<T: DeserializeOwned, P: Serialize>(
+    pub(crate) fn post_raw<T: DeserializeOwned, P: Serialize>(
         &mut self,
         url: &str,
         params: &P,
     ) -> Result<T, ApiError> {
-        self.check_expired().await?;
-        let req = self.client.post(url)
-            .header("Accept", "application/json")
-            .header("Authorization", &self.auth_header)
-            .json(params);
+        self.check_expired()?;
+        let req = ureq::post(url)
+            .set("Accept", "application/json")
+            .set("Authorization", &self.auth_header);
 
-        let mut resp = req
-            .try_clone().unwrap()
-            .send().await?;
+        let mut resp = req.clone().send_json(params)?;
 
         if resp.status() == 401 {
-            self.refresh_token().await?;
-            resp = req.send().await?
+            self.refresh_token()?;
+            resp = req.send_json(params)?
         }
 
-        resp.into_json_with_error().await
+        resp.into_json_with_error()
     }
 }
 
@@ -593,12 +584,12 @@ pub struct MeterReading {
 
 impl Site {
     /// Read all energy meters from the given site
-    pub async fn lifetime_energy(&self, ctx: &mut Context) -> Result<Vec<MeterReading>, ApiError> {
-        ctx.get(&format!("sites/{}/energy", self.id)).await
+    pub fn lifetime_energy(&self, ctx: &mut Context) -> Result<Vec<MeterReading>, ApiError> {
+        ctx.get(&format!("sites/{}/energy", self.id))
     }
 
-    pub async fn details(&self, ctx: &mut Context) -> Result<SiteDetails, ApiError> {
-        ctx.get(&format!("sites/{}", self.id)).await
+    pub fn details(&self, ctx: &mut Context) -> Result<SiteDetails, ApiError> {
+        ctx.get(&format!("sites/{}", self.id))
     }
 }
 
@@ -607,63 +598,63 @@ impl Circuit {
         format!("sites/{}/circuits/{}/dynamicCurrent", self.site_id, self.id)
     }
 
-    pub async fn dynamic_current(&self, ctx: &mut Context) -> Result<Triphase, ApiError> {
-        ctx.circuit_dynamic_current(self.site_id, self.id).await
+    pub fn dynamic_current(&self, ctx: &mut Context) -> Result<Triphase, ApiError> {
+        ctx.circuit_dynamic_current(self.site_id, self.id)
     }
 
-    pub async fn set_dynamic_current(
+    pub fn set_dynamic_current(
         &self,
         ctx: &mut Context,
         current: SetCurrent,
     ) -> Result<(), ApiError> {
-        ctx.post(&self.dynamic_current_path(), &current).await
+        ctx.post(&self.dynamic_current_path(), &current)
     }
 }
 
 impl Charger {
     /// Enable "smart charging" on the charger. This just turns the LED blue, and disables basic charging plans.
-    pub async fn enable_smart_charging(&self, ctx: &mut Context) -> Result<(), ApiError> {
+    pub fn enable_smart_charging(&self, ctx: &mut Context) -> Result<(), ApiError> {
         let url = format!("chargers/{}/commands/smart_charging", &self.id);
-        ctx.post(&url, &()).await
+        ctx.post(&url, &())
     }
 
     /// Read the state of a charger
-    pub async fn state(&self, ctx: &mut Context) -> Result<ChargerState, ApiError> {
+    pub fn state(&self, ctx: &mut Context) -> Result<ChargerState, ApiError> {
         let url = format!("chargers/{}/state", self.id);
-        ctx.get(&url).await
+        ctx.get(&url)
     }
 
     /// Read info about the ongoing charging session
-    pub async fn ongoing_session(&self, ctx: &mut Context) -> Result<Option<ChargingSession>, ApiError> {
-        ctx.maybe_get(&format!("chargers/{}/sessions/ongoing", &self.id)).await
+    pub fn ongoing_session(&self, ctx: &mut Context) -> Result<Option<ChargingSession>, ApiError> {
+        ctx.maybe_get(&format!("chargers/{}/sessions/ongoing", &self.id))
     }
 
     /// Read info about the last charging session (not including ongoing one)
-    pub async fn latest_session(&self, ctx: &mut Context) -> Result<Option<ChargingSession>, ApiError> {
-        ctx.maybe_get(&format!("chargers/{}/sessions/latest", &self.id)).await
+    pub fn latest_session(&self, ctx: &mut Context) -> Result<Option<ChargingSession>, ApiError> {
+        ctx.maybe_get(&format!("chargers/{}/sessions/latest", &self.id))
     }
 
-    async fn command(&self, ctx: &mut Context, command: &str) -> Result<CommandReply, ApiError> {
-        ctx.post(&format!("chargers/{}/commands/{}", self.id, command), &()).await
+    fn command(&self, ctx: &mut Context, command: &str) -> Result<CommandReply, ApiError> {
+        ctx.post(&format!("chargers/{}/commands/{}", self.id, command), &())
     }
 
-    pub async fn start(&self, ctx: &mut Context) -> Result<(), ApiError> {
-        self.command(ctx, "start_charging").await?;
+    pub fn start(&self, ctx: &mut Context) -> Result<(), ApiError> {
+        self.command(ctx, "start_charging")?;
         Ok(())
     }
 
-    pub async fn pause(&self, ctx: &mut Context) -> Result<(), ApiError> {
-        self.command(ctx, "pause_charging").await?;
+    pub fn pause(&self, ctx: &mut Context) -> Result<(), ApiError> {
+        self.command(ctx, "pause_charging")?;
         Ok(())
     }
 
-    pub async fn resume(&self, ctx: &mut Context) -> Result<(), ApiError> {
-        self.command(ctx, "resume_charging").await?;
+    pub fn resume(&self, ctx: &mut Context) -> Result<(), ApiError> {
+        self.command(ctx, "resume_charging")?;
         Ok(())
     }
 
-    pub async fn stop(&self, ctx: &mut Context) -> Result<(), ApiError> {
-        self.command(ctx, "stop_charging").await?;
+    pub fn stop(&self, ctx: &mut Context) -> Result<(), ApiError> {
+        self.command(ctx, "stop_charging")?;
         Ok(())
     }
 }
@@ -675,15 +666,11 @@ mod test {
     use super::Context;
     #[test]
     fn token_save() {
-
-        let client = reqwest::Client::new();
-
         let ctx = Context {
             auth_header: "Bearer aaaaaaa0".to_owned(),
             refresh_token: "abcdef".to_owned(),
             token_expiration: Instant::now() + Duration::from_secs(1234),
             on_refresh: None,
-            client: client.clone(),
         };
 
         let saved = ctx.save();

src/observation.rs

@@ -1,9 +1,12 @@
-use serde::{de::{DeserializeOwned, IntoDeserializer}, Deserialize};
-use serde_json::json;
+use serde::{
+    de::{DeserializeOwned, IntoDeserializer},
+    Deserialize,
+};
 use serde_repr::Deserialize_repr;
 use std::num::{ParseFloatError, ParseIntError};
 use thiserror::Error;
 use tracing::info;
+use ureq::json;
 
 use crate::{
     api::{ChargerOpMode, Context, OutputPhase, UtcDateTime},
@@ -83,16 +86,16 @@ pub enum ParseError {
 impl ObservationData {
     fn from_dynamic(value: String, data_type: DataType) -> Result<ObservationData, ParseError> {
         Ok(match data_type {
-            DataType::Boolean => ObservationData::Boolean(
-                match &*value {
-                    "False"|"false" => { false },
-                    "True"|"true" => { true }
-                    other => other
-                    .parse::<i64>()
-                    .map_err(move |e| ParseError::Integer(value, e))?
-                    != 0,
+            DataType::Boolean => ObservationData::Boolean(match &*value {
+                "False" | "false" => false,
+                "True" | "true" => true,
+                other => {
+                    other
+                        .parse::<i64>()
+                        .map_err(move |e| ParseError::Integer(value, e))?
+                        != 0
                 }
-            ),
+            }),
             DataType::Double => ObservationData::Double(
                 value
                     .parse()
@@ -226,7 +229,8 @@ fn op_mode_from_int(mode: i64) -> ChargerOpMode {
 }
 
 fn deserialize_i64<T: DeserializeOwned>(value: i64) -> Option<T> {
-    T::deserialize(<i64 as IntoDeserializer<serde::de::value::Error>>::into_deserializer(value)).ok()
+    T::deserialize(<i64 as IntoDeserializer<serde::de::value::Error>>::into_deserializer(value))
+        .ok()
 }
 
 impl Observation {
@@ -325,17 +329,17 @@ struct ProductUpdate {
 }
 
 impl Stream {
-    pub async fn from_context(ctx: &mut Context) -> Result<Self, NegotiateError> {
+    pub fn from_context(ctx: &mut Context) -> Result<Self, NegotiateError> {
         Ok(Self {
-            inner: signalr::Stream::from_ws(crate::stream::Stream::open(ctx).await?),
+            inner: signalr::Stream::from_ws(crate::stream::Stream::open(ctx)?),
         })
     }
 
-    pub async fn recv(&mut self) -> Result<Event, ObservationError> {
+    pub fn recv(&mut self) -> Result<Event, ObservationError> {
         use signalr::Message::*;
         let de = |msg| -> Result<Event, ObservationError> { Err(ObservationError::Protocol(msg)) };
         loop {
-            let msg = self.inner.recv().await?;
+            let msg = self.inner.recv()?;
             match &msg {
                 Ping => continue,
                 Empty | InvocationResult { .. } => info!("Skipped message: {msg:?}"),
@@ -351,9 +355,9 @@ impl Stream {
             }
         }
     }
-    pub async fn subscribe(&mut self, id: &str) -> Result<(), tungstenite::Error> {
+    pub fn subscribe(&mut self, id: &str) -> Result<(), tungstenite::Error> {
         self.inner
-            .invoke("SubscribeWithCurrentState", json!([id, true])).await
+            .invoke("SubscribeWithCurrentState", json!([id, true]))
     }
 }
 

src/signalr.rs

@@ -113,9 +113,9 @@ impl Stream {
         Self { ws, buffer: vec![] }
     }
 
-    pub async fn recv(&mut self) -> Result<Message, StreamError> {
+    pub fn recv(&mut self) -> Result<Message, StreamError> {
         while self.buffer.is_empty() {
-            self.buffer = self.ws.recv().await?;
+            self.buffer = self.ws.recv()?;
             self.buffer.reverse();
         }
 
@@ -123,7 +123,7 @@ impl Stream {
         Ok(Message::from_json(json)?)
     }
 
-    pub async fn invoke(
+    pub fn invoke(
         &mut self,
         target: &str,
         args: serde_json::Value,
@@ -131,6 +131,6 @@ impl Stream {
         self.ws.send(json!( { "arguments": args,
                                   "invocationId": "0",
                                   "target": target,
-                                  "type": 1} )).await
+                                  "type": 1} ))
     }
 }

src/stream.rs

@@ -1,11 +1,9 @@
 use super::api::{ApiError, Context};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
-use tokio::net::TcpStream;
+use std::net::TcpStream;
 use thiserror::Error;
-//use tungstenite::{stream::MaybeTlsStream, Message, WebSocket};
-use tokio_tungstenite::{MaybeTlsStream, tungstenite::Message, WebSocketStream};
-use futures_util::{SinkExt,StreamExt};
+use tungstenite::{stream::MaybeTlsStream, Message, WebSocket};
 
 const STREAM_API_NEGOTIATION_URL: &str =
     "https://streams.easee.com/hubs/products/negotiate?negotiateVersion=1";
@@ -38,18 +36,15 @@ pub enum RecvError {
 
     #[error("WS error: {0}")]
     TungsteniteError(#[from] tungstenite::Error),
-
-    #[error("End of stream")]
-    EndOfStream,
 }
 
 pub struct Stream {
-    sock: WebSocketStream<MaybeTlsStream<TcpStream>>,
+    sock: WebSocket<MaybeTlsStream<TcpStream>>,
 }
 
 impl Stream {
-    pub async fn open(ctx: &mut Context) -> Result<Stream, NegotiateError> {
-        let r: NegotiateResponse = ctx.post_raw(STREAM_API_NEGOTIATION_URL, &()).await?;
+    pub fn open(ctx: &mut Context) -> Result<Stream, NegotiateError> {
+        let r: NegotiateResponse = ctx.post_raw(STREAM_API_NEGOTIATION_URL, &())?;
 
         let token = ctx.auth_token();
         let wss_url = format!(
@@ -57,8 +52,7 @@ impl Stream {
             WSS_URL, r.connection_token, token
         );
 
-        let resp = tokio_tungstenite::connect_async(wss_url).await;
-        //let resp = tungstenite::client::connect(&wss_url);
+        let resp = tungstenite::client::connect(&wss_url);
 
         if let Err(tungstenite::Error::Http(he)) = &resp {
             eprintln!(
@@ -68,20 +62,19 @@ impl Stream {
         }
 
         let mut stream = Stream { sock: resp?.0 };
-        stream.send(json!({ "protocol": "json", "version": 1 })).await?;
+        stream.send(json!({ "protocol": "json", "version": 1 }))?;
 
         Ok(stream)
     }
 
-    pub async fn send<T: Serialize>(&mut self, msg: T) -> Result<(), tungstenite::Error> {
+    pub fn send<T: Serialize>(&mut self, msg: T) -> Result<(), tungstenite::Error> {
         let mut msg = serde_json::to_string(&msg).unwrap();
         msg.push('\x1E');
-        self.sock.send(Message::Text(msg)).await
+        self.sock.send(Message::Text(msg))
     }
 
-    pub async fn recv(&mut self) -> Result<Vec<serde_json::Value>, RecvError> {
-        let msg = self.sock.next().await
-            .ok_or(RecvError::EndOfStream)??;
+    pub fn recv(&mut self) -> Result<Vec<serde_json::Value>, RecvError> {
+        let msg = self.sock.read()?;
         let Message::Text(txt) = msg else {
             return Err(RecvError::BadMessageType);
         };