overhaul

Author: refcell

Cargo.toml

@@ -22,3 +22,11 @@ ethers = { version = "0.17.0", features = [ "ws", "rustls" ] }
 serde_json = { version = "1.0.64", default-features = false, features = ["raw_value"] }
 serde = "1.0.144"
 actix-rt = "2.7.0"
+soketto = "0.7.1"
+tokio = { version = "1", features = ["full"] }
+tokio-util = { version = "0.6", features = ["compat"] }
+tokio-stream = { version = "0.1", features = ["net"] }
+futures = { default-features = false, features = ["bilock", "std", "unstable"], version = "0.3.1" }
+tracing = "0.1.36"
+tracing-subscriber = "0.3.15"
+uuid = { version = "1.1.2", features = [ "serde" ] }

README.md

@@ -17,19 +17,77 @@ alchemy_rs = "0.1.0"
 
 ## Usage
 
-// TODO
+[alchemy-rs](https://github.com/abigger87/alchemy-rs) is a minimal ethers-rs wrapper for the Alchemy API built in pure rust.
+
+The [AlchemyManager](src/manager.rs) is the main entry point for interacting with the Alchemy API. It is initializable with an Alchemy API key and a [Chain](https://docs.rs/ethers/latest/ethers/types/enum.Chain.html). Alchemy supports the following chains: ...
+
 
 
 ## Examples
 
-```rust
+Listening to pending transactions using alchemy's `alchemy_pendingTransactions` method is demonstrated below.
 
+```rust
+use std::str::FromStr;
+use std::env;
+
+use alchemy_rs::prelude::*;
+
+async {
+    // Read an alchemy websocket api key from the `ALCHEMY_API_KEY` environment variable
+    let api_key = env::var("ALCHEMY_API_KEY").expect("ALCHEMY_API_KEY must be set");
+
+    // Create the AlchemyManager
+    let mut manager = AlchemyManager::new(&format!("wss://eth-mainnet.g.alchemy.com/v2/{}", api_key), None);
+
+    // Connect to the websocket
+    let _ = manager.connect().await.unwrap();
+
+    // Listen to _pending_ transactions to the USDT address on mainnet
+    // (there should be a lot of these!)
+    let usdt_address = Address::from_str("dac17f958d2ee523a2206206994597c13d831ec7").unwrap();
+
+    // Try to subscribe to pending transactions
+    let sub_id = match manager.subscribe(Some(usdt_address), None).await {
+        Ok(id) => id,
+        Err(e) => {
+            println!("Error subscribing to pending transactions: {:?}", e);
+            return;
+        }
+    };
+
+    // Now we can grab items from the stream
+    let item: AlchemySocketMessageResponse;
+    loop {
+        match manager.receive(sub_id).await {
+            Ok(i) => {
+                item = i;
+                break;
+            },
+            Err(e) => {
+                println!("Error receiving item: {:?}", e);
+                return;
+            }
+        }
+    }
+
+   // Print the next item
+   println!("Received pending transaction from the stream: {:?}", item);
+};
 ```
 
 
+## Safety
+
+> **Warning**
+>
+> This is **experimental software** and is provided on an "as is" and "as available" basis.
+> Expect rapid iteration and **use at your own risk**.
+
+
 ## License
 
-[AGPL-3.0-only](https://github.com/abigger87/alchemy-rs/blob/master/LICENSE)
+[MIT](https://github.com/abigger87/alchemy-rs/blob/master/LICENSE), but go crazy :P
 
 
 ## Acknowledgements

src/connectors/errors.rs

@@ -0,0 +1,38 @@
+use ethers::providers::ProviderError;
+use soketto::handshake;
+
+/// An Alchemy Websocket Connection Error
+#[derive(Debug)]
+pub enum AlchemyConnectionError {
+    /// A Provider Connection Error
+    ProviderError(ProviderError),
+    /// Raw Websocket TcpStream Error
+    RawStreamError(std::io::Error),
+    /// A Raw Websocket Connection Error
+    RawSocketError(soketto::handshake::Error),
+    /// A Raw Websocket Handshake Error
+    RawHandshakeError(handshake::Error),
+    /// Deserialization Error
+    Deserialization(serde_json::Error),
+    /// Received an unexpected response type
+    UnexpectedResponseType,
+    /// Missing the websocket channel sender
+    MissingSender,
+    /// Missing the websocket channel receiver
+    MissingReceiver,
+    /// Connection Closed
+    Closed,
+    /// No websocket connection established yet
+    MissingConnection,
+    /// Sending a message to the websocket channel failed
+    SendFailed(soketto::connection::Error),
+    /// Flushing the websocket channel failed
+    FlushFailed(soketto::connection::Error),
+
+    /// Some soketto websocket error
+    SomeError(soketto::connection::Error),
+    /// The method is unimplemented
+    Unimplemented,
+    /// The text response could not be parsed as a string
+    InvalidTextString,
+}

src/connectors/mod.rs

@@ -0,0 +1,48 @@
+//! Alchemy Connectors
+
+/// An ethers provider connector
+pub mod provider;
+
+/// A raw websocket connector
+pub mod raw;
+
+/// Common Errors
+pub mod errors;
+
+/// Re-export a prelude
+pub mod prelude {
+    pub use super::{errors::*, provider::*, raw::*, *};
+}
+
+/// An alchemy api connection manager
+#[derive(Debug)]
+pub enum AlchemyConnector {
+    /// An ethers-rs websocket [Provider](ethers::providers::Provider) for alchemy
+    Provider(Option<provider::EthersWsProvider>),
+    /// A Raw, Persistent Websocket Connection to the Alchemy API using [soketto](https://docs.rs/soketto/latest/soketto/)
+    Raw(Option<raw::RawAlchemyConnection>),
+}
+
+/// The type of alchemy api websocket connection
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum AlchemyConnectorType {
+    /// An ethers-rs websocket [Provider](ethers::providers::Provider) for alchemy
+    Provider,
+    /// A Raw, Persistent Websocket Connection to the Alchemy API using [soketto](https://docs.rs/soketto/latest/soketto/)
+    Raw,
+}
+
+impl Default for AlchemyConnectorType {
+    fn default() -> Self {
+        AlchemyConnectorType::Raw
+    }
+}
+
+impl From<AlchemyConnectorType> for AlchemyConnector {
+    fn from(t: AlchemyConnectorType) -> Self {
+        match t {
+            AlchemyConnectorType::Provider => AlchemyConnector::Provider(None),
+            AlchemyConnectorType::Raw => AlchemyConnector::Raw(None),
+        }
+    }
+}

src/connectors/provider.rs

@@ -0,0 +1,36 @@
+use ethers::prelude::*;
+
+use super::errors::AlchemyConnectionError;
+
+/// An ethers-rs websocket [Provider](ethers::providers::Provider) for alchemy
+#[derive(Debug, Default, Clone)]
+pub struct EthersWsProvider {
+    /// The ethers-rs websocket [Provider](ethers::providers::Provider)
+    pub provider: Option<Provider<Ws>>,
+}
+
+impl EthersWsProvider {
+    /// Create a new ethers-rs websocket [Provider](ethers::providers::Provider) for alchemy
+    pub fn new() -> Self {
+        Self { provider: None }
+    }
+
+    /// Connect to the websocket provider
+    pub async fn connect(&mut self, url: &str) -> Result<(), AlchemyConnectionError> {
+        match Provider::connect(String::from(url)).await {
+            Ok(p) => {
+                self.provider = Some(p);
+                Ok(())
+            }
+            Err(e) => Err(AlchemyConnectionError::ProviderError(e)),
+        }
+    }
+}
+
+impl From<Provider<Ws>> for EthersWsProvider {
+    fn from(provider: Provider<Ws>) -> Self {
+        Self {
+            provider: Some(provider),
+        }
+    }
+}

src/connectors/raw.rs

@@ -0,0 +1,81 @@
+use futures::io::{BufReader, BufWriter};
+use soketto::handshake;
+use tokio::net::TcpStream;
+use tokio_util::compat::{Compat, TokioAsyncReadCompatExt};
+
+use super::errors::AlchemyConnectionError;
+
+/// A Raw, Persistent Websocket Connection to the Alchemy API using [soketto](https://docs.rs/soketto/latest/soketto/)
+///
+/// ## Alchemy
+///
+/// The Alchemy Websocket API allows you to interactively demo the endpoints.
+///
+/// Simply install a websocket shell command and connect to the demo endpoint:
+///
+/// ```sh
+/// $ wscat -c wss://eth-mainnet.ws.alchemyapi.io/ws/demo
+///
+///  // create subscription
+/// > {"id": 1, "method": "eth_subscribe", "params": ["newHeads"]}
+/// < {"jsonrpc":"2.0","id":1,"result":"0xcd0c3e8af590364c09d0fa6a1210faf5"}
+///
+/// // incoming notifications
+/// < {"jsonrpc":"2.0","method":"eth_subscription","params":{"subscription":"0xcd0c3e8af590364c09d0fa6a1210faf5","result":{"difficulty":"0xd9263f42a87",<...>, "uncles":[]}}}
+/// < {"jsonrpc":"2.0","method":"eth_subscription","params":{"subscription":"0xcd0c3e8af590364c09d0fa6a1210faf5","result":{"difficulty":"0xd90b1a7ad02", <...>, "uncles":["0x80aacd1ea4c9da32efd8c2cc9ab38f8f70578fcd46a1a4ed73f82f3e0957f936"]}}}
+///
+/// // cancel subscription
+/// > {"id": 1, "method": "eth_unsubscribe", "params": ["0xcd0c3e8af590364c09d0fa6a1210faf5"]}
+/// < {"jsonrpc":"2.0","id":1,"result":true}
+/// ```
+///
+#[derive(Debug, Default)]
+pub struct RawAlchemyConnection {
+    // / The websocket connection
+    // pub connection: Option<soketto::handshake::Client<'a, BufReader<BufWriter<Compat<TcpStream>>>>>,
+    /// The websocket client sender after building
+    pub sender: Option<soketto::Sender<BufReader<BufWriter<Compat<TcpStream>>>>>,
+    /// The websocket client receiver after building
+    pub receiver: Option<soketto::Receiver<BufReader<BufWriter<Compat<TcpStream>>>>>,
+}
+
+impl RawAlchemyConnection {
+    /// Create a new RawAlchemyConnection
+    pub fn new() -> Self {
+        Self {
+            sender: None,
+            receiver: None,
+        }
+    }
+
+    /// Connect to the sokettot websocket
+    pub async fn connect(&mut self, url: &'_ str) -> Result<(), AlchemyConnectionError> {
+        // Create the socket connection
+        let socket = match tokio::net::TcpStream::connect(url).await {
+            Ok(s) => s,
+            Err(e) => return Err(AlchemyConnectionError::RawStreamError(e)),
+        };
+
+        // Create the client connection
+        let compatible_socket = BufReader::new(BufWriter::new(socket.compat()));
+        let mut client = handshake::Client::new(compatible_socket, url, "");
+
+        // Handshake the connection
+        match client.handshake().await {
+            Ok(sr) => {
+                tracing::info!("Got handshake response: {:?}", sr);
+                tracing::debug!(
+                    "Expecting server response: {:?}",
+                    handshake::ServerResponse::Accepted { protocol: None }
+                );
+            }
+            Err(e) => return Err(AlchemyConnectionError::RawHandshakeError(e)),
+        }
+
+        let (sender, receiver) = client.into_builder().finish();
+        self.sender = Some(sender);
+        self.receiver = Some(receiver);
+
+        Ok(())
+    }
+}

src/lib.rs

@@ -3,9 +3,14 @@
 #![forbid(unsafe_code)]
 #![forbid(where_clauses_object_safety)]
 #![deny(rustdoc::broken_intra_doc_links)]
-
 #![doc=include_str!("../README.md")]
 
+/// Refactored Websocket Connectors for the Alchemy Manager
+pub mod connectors;
+
+/// Refactored Alchemy Websocket Messages
+pub mod messages;
+
 /// Alchemy Manager
 pub mod manager;
 
@@ -17,5 +22,8 @@ pub mod types;
 
 /// A prelude of commonly used alchemy-rs items
 pub mod prelude {
-    pub use super::{manager::*, types::*, wrapper::*};
+    pub use super::{manager::*, messages::prelude::*, types::*, wrapper::*};
+
+    // Re-export ethers-rs prelude
+    pub use ethers::prelude::*;
 }