Update client to work asynchronously

 - Use reqwest's async client
 - Convert items_iter() to a futures::Stream
 - make Mastodon client an Arc smart pointer, removing the need for OwnedPage.
 - remove MastodonClient and HttpSender traits; these can be re-added once async trait fns are stabilized
 - make EventStream a futures::Stream

20 files changed, 879 insertions, 1077 deletions

diff --git a/Cargo.toml b/Cargo.toml
index 7eec89a..db6978b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,97 +1,70 @@
-# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
-#
-# When uploading crates to the registry Cargo will automatically
-# "normalize" Cargo.toml files for maximal compatibility
-# with all versions of Cargo and also rewrite `path` dependencies
-# to registry (e.g., crates.io) dependencies
-#
-# If you believe there's an error in this file please file an
-# issue against the rust-lang/cargo repository. If you're
-# editing this file be aware that the upstream Cargo.toml
-# will likely look very different (and much more reasonable)
-
 [package]
 name = "elefren"
 version = "0.23.0"
-authors = ["Aaron Power <theaaronepower@gmail.com>", "Paul Woolcock <paul@woolcock.us>"]
+authors = ["Aaron Power <theaaronepower@gmail.com>", "Paul Woolcock <paul@woolcock.us>", "D. Scott Boggs <scott@tams.tech>"]
 description = "A wrapper around the Mastodon API."
 readme = "README.md"
 keywords = ["api", "web", "social", "mastodon", "wrapper"]
 categories = ["web-programming", "web-programming::http-client", "api-bindings"]
 license = "MIT/Apache-2.0"
 edition = "2021"
-# TODO setup repo
-# repository = "https://github.com/pwoolcoc/elefren.git"
+repository = "https://github.com/dscottboggs/elefren.git"
 
 [package.metadata.docs.rs]
 features = ["all"]
+
+[dependencies]
+futures = "0.3.25"
+doc-comment = "0.3"
+log = "0.4"
+serde_json = "1"
+serde_qs = "0.4.5"
+serde_urlencoded = "0.6.1"
+tap-reader = "1"
+tungstenite = "0.18"
+url = "1"
+# Provides parsing for the link header in get_links() in page.rs
+hyper-old-types = "0.11.0"
+
 [dependencies.chrono]
 version = "0.4"
 features = ["serde"]
 
-[dependencies.doc-comment]
-version = "0.3"
-
 [dependencies.envy]
 version = "0.4"
 optional = true
 
-# Provides parsing for the link header in get_links() in page.rs
-[dependencies.hyper-old-types]
-version = "0.11.0"
-
 [dependencies.isolang]
 version = "2.2"
 features = ["serde"]
 
-[dependencies.log]
-version = "0.4"
-
 [dependencies.reqwest]
-version = "0.9"
-default-features = false
+version = "0.11"
+features = ["multipart", "json"]
 
 [dependencies.serde]
 version = "1"
 features = ["derive"]
 
-[dependencies.serde_json]
-version = "1"
-
-[dependencies.serde_qs]
-version = "0.4.5"
-
-[dependencies.serde_urlencoded]
-version = "0.6.1"
-
-[dependencies.tap-reader]
-version = "1"
-
 [dependencies.toml]
 version = "0.5"
 optional = true
 
-[dependencies.tungstenite]
-version = "0.10.1"
-
-[dependencies.url]
-version = "1"
-
-[dev-dependencies.indoc]
-version = "1.0"
-
-[dev-dependencies.pretty_env_logger]
-version = "0.3.0"
-
-[dev-dependencies.skeptic]
-version = "0.13"
-
-[dev-dependencies.tempfile]
-version = "3"
+[dev-dependencies]
+tokio-test = "0.4.2"
+futures-util = "0.3.25"
+indoc = "1.0"
+pretty_env_logger = "0.3.0"
+skeptic = "0.13"
+tempfile = "3"
 
 [build-dependencies.skeptic]
 version = "0.13"
 
+[dev-dependencies.tokio]
+version = "1.22.0"
+features = ["rt-multi-thread", "macros"]
+
 [features]
 all = ["toml", "json", "env"]
 default = ["reqwest/default-tls"]
diff --git a/examples/follow_profile.rs b/examples/follow_profile.rs
index 84b2998..6d50327 100644
--- a/examples/follow_profile.rs
+++ b/examples/follow_profile.rs
@@ -5,7 +5,7 @@ extern crate pretty_env_logger;
 extern crate elefren;
 mod register;
 
-use register::MastodonClient;
+use register::Mastodon;
 use std::error;
 
 #[cfg(feature = "toml")]
diff --git a/examples/follows_me.rs b/examples/follows_me.rs
index a78ceb7..c491994 100644
--- a/examples/follows_me.rs
+++ b/examples/follows_me.rs
@@ -4,7 +4,7 @@
 extern crate pretty_env_logger;
 mod register;
 
-use register::MastodonClient;
+use register::Mastodon;
 use std::error;
 
 #[cfg(feature = "toml")]
diff --git a/examples/home_timeline.rs b/examples/home_timeline.rs
index 84531a6..ae279a9 100644
--- a/examples/home_timeline.rs
+++ b/examples/home_timeline.rs
@@ -4,7 +4,7 @@
 extern crate pretty_env_logger;
 mod register;
 
-use register::MastodonClient;
+use register::Mastodon;
 use std::error;
 
 #[cfg(feature = "toml")]
diff --git a/examples/print_your_profile.rs b/examples/print_your_profile.rs
index cbcd372..cae9f6e 100644
--- a/examples/print_your_profile.rs
+++ b/examples/print_your_profile.rs
@@ -5,7 +5,7 @@ extern crate pretty_env_logger;
 extern crate elefren;
 mod register;
 
-use register::MastodonClient;
+use register::Mastodon;
 use std::error;
 
 #[cfg(feature = "toml")]
diff --git a/examples/search.rs b/examples/search.rs
index 28ef8df..4bd2756 100644
--- a/examples/search.rs
+++ b/examples/search.rs
@@ -4,7 +4,7 @@
 extern crate pretty_env_logger;
 mod register;
 
-use register::MastodonClient;
+use register::Mastodon;
 use std::error;
 
 #[cfg(feature = "toml")]
diff --git a/examples/upload_photo.rs b/examples/upload_photo.rs
index a2e6688..8ce184c 100644
--- a/examples/upload_photo.rs
+++ b/examples/upload_photo.rs
@@ -5,7 +5,7 @@ extern crate pretty_env_logger;
 extern crate elefren;
 mod register;
 
-use register::MastodonClient;
+use register::Mastodon;
 use std::error;
 
 #[cfg(feature = "toml")]
diff --git a/src/entities/itemsiter.rs b/src/entities/itemsiter.rs
index a8395e7..46511d8 100644
--- a/src/entities/itemsiter.rs
+++ b/src/entities/itemsiter.rs
@@ -1,27 +1,36 @@
-use crate::{http_send::HttpSend, page::Page};
+use futures::{stream::unfold, Stream};
+
+use crate::page::Page;
 use serde::Deserialize;
 
 /// Abstracts away the `next_page` logic into a single stream of items
 ///
-/// ```no_run
+/// ```no_run,async
 /// use elefren::prelude::*;
-/// let data = Data::default();
-/// let client = Mastodon::from(data);
-/// let statuses = client.statuses("user-id", None).unwrap();
-/// for status in statuses.items_iter() {
-///     // do something with `status`
-/// }
+/// use futures::stream::StreamExt;
+/// use futures_util::pin_mut;
+///
+/// tokio_test::block_on(async {
+///     let data = Data::default();
+///     let client = Mastodon::from(data);
+///     let statuses = client.statuses("user-id", None).await.unwrap().items_iter();
+///     statuses.for_each(|status| async move {
+///         // Do something with the status
+///     }).await;
+/// })
 /// ```
+///
+/// See documentation for `futures::Stream::StreamExt` for available methods.
 #[derive(Debug, Clone)]
-pub(crate) struct ItemsIter<'a, T: Clone + for<'de> Deserialize<'de>, H: 'a + HttpSend> {
-    page: Page<'a, T, H>,
+pub(crate) struct ItemsIter<T: Clone + for<'de> Deserialize<'de>> {
+    page: Page<T>,
     buffer: Vec<T>,
     cur_idx: usize,
     use_initial: bool,
 }
 
-impl<'a, T: Clone + for<'de> Deserialize<'de>, H: HttpSend> ItemsIter<'a, T, H> {
-    pub(crate) fn new(page: Page<'a, T, H>) -> ItemsIter<'a, T, H> {
+impl<'a, T: Clone + for<'de> Deserialize<'de>> ItemsIter<T> {
+    pub(crate) fn new(page: Page<T>) -> ItemsIter<T> {
         ItemsIter {
             page,
             buffer: vec![],
@@ -34,8 +43,8 @@ impl<'a, T: Clone + for<'de> Deserialize<'de>, H: HttpSend> ItemsIter<'a, T, H>
         self.buffer.is_empty() || self.cur_idx == self.buffer.len()
     }
 
-    fn fill_next_page(&mut self) -> Option<()> {
-        let items = if let Ok(items) = self.page.next_page() {
+    async fn fill_next_page(&mut self) -> Option<()> {
+        let items = if let Ok(items) = self.page.next_page().await {
             items
         } else {
             return None;
@@ -51,33 +60,39 @@ impl<'a, T: Clone + for<'de> Deserialize<'de>, H: HttpSend> ItemsIter<'a, T, H>
             None
         }
     }
-}
 
-impl<'a, T: Clone + for<'de> Deserialize<'de>, H: HttpSend> Iterator for ItemsIter<'a, T, H> {
-    type Item = T;
-
-    fn next(&mut self) -> Option<Self::Item> {
-        if self.use_initial {
-            if self.page.initial_items.is_empty() || self.cur_idx == self.page.initial_items.len() {
-                return None;
-            }
-            let idx = self.cur_idx;
-            if self.cur_idx == self.page.initial_items.len() - 1 {
-                self.cur_idx = 0;
-                self.use_initial = false;
-            } else {
-                self.cur_idx += 1;
-            }
-            Some(self.page.initial_items[idx].clone())
-        } else {
-            if self.need_next_page() {
-                if self.fill_next_page().is_none() {
+    pub(crate) fn stream(self) -> impl Stream<Item = T> {
+        unfold(self, |mut this| async move {
+            if this.use_initial {
+                if this.page.initial_items.is_empty()
+                    || this.cur_idx == this.page.initial_items.len()
+                {
                     return None;
                 }
+                let idx = this.cur_idx;
+                if this.cur_idx == this.page.initial_items.len() - 1 {
+                    this.cur_idx = 0;
+                    this.use_initial = false;
+                } else {
+                    this.cur_idx += 1;
+                }
+                let item = this.page.initial_items[idx].clone();
+                // let item = Box::pin(item);
+                // pin_mut!(item);
+                Some((item, this))
+            } else {
+                if this.need_next_page() {
+                    if this.fill_next_page().await.is_none() {
+                        return None;
+                    }
+                }
+                let idx = this.cur_idx;
+                this.cur_idx += 1;
+                let item = this.buffer[idx].clone();
+                // let item = Box::pin(item);
+                // pin_mut!(item);
+                Some((item, this))
             }
-            let idx = self.cur_idx;
-            self.cur_idx += 1;
-            Some(self.buffer[idx].clone())
-        }
+        })
     }
 }
diff --git a/src/errors.rs b/src/errors.rs
index b8b4c18..754d274 100644
--- a/src/errors.rs
+++ b/src/errors.rs
@@ -1,4 +1,4 @@
-use std::{error, fmt, io::Error as IoError};
+use std::{error, fmt, io::Error as IoError, num::TryFromIntError};
 
 #[cfg(feature = "env")]
 use envy::Error as EnvyError;
@@ -12,7 +12,7 @@ use serde_urlencoded::ser::Error as UrlEncodedError;
 use tomlcrate::de::Error as TomlDeError;
 #[cfg(feature = "toml")]
 use tomlcrate::ser::Error as TomlSerError;
-use tungstenite::error::Error as WebSocketError;
+use tungstenite::{error::Error as WebSocketError, Message as WebSocketMessage};
 use url::ParseError as UrlError;
 
 /// Convience type over `std::result::Result` with `Error` as the error type.
@@ -64,6 +64,14 @@ pub enum Error {
     SerdeQs(SerdeQsError),
     /// WebSocket error
     WebSocket(WebSocketError),
+    /// An integer conversion was attempted, but the value didn't fit into the
+    /// target type.
+    ///
+    /// At the time of writing, this can only be triggered when a file is
+    /// larger than the system's usize allows.
+    IntConversion(TryFromIntError),
+    /// A stream message was received that wasn't recognized
+    UnrecognizedStreamMessage(WebSocketMessage),
     /// Other errors
     Other(String),
 }
@@ -93,12 +101,13 @@ impl error::Error for Error {
             Error::Envy(ref e) => e,
             Error::SerdeQs(ref e) => e,
             Error::WebSocket(ref e) => e,
-
+            Error::IntConversion(ref e) => e,
             Error::Client(..) | Error::Server(..) => return None,
             Error::ClientIdRequired => return None,
             Error::ClientSecretRequired => return None,
             Error::AccessTokenRequired => return None,
             Error::MissingField(_) => return None,
+            Error::UnrecognizedStreamMessage(_) => return None,
             Error::Other(..) => return None,
         })
     }
@@ -122,7 +131,7 @@ impl fmt::Display for ApiError {
 impl error::Error for ApiError {}
 
 macro_rules! from {
-    ($($(#[$met:meta])* $typ:ident, $variant:ident,)*) => {
+    ($($(#[$met:meta])* $typ:ident => $variant:ident,)*) => {
         $(
             $(#[$met])*
             impl From<$typ> for Error {
@@ -136,20 +145,22 @@ macro_rules! from {
 }
 
 from! {
-    HttpError, Http,
-    IoError, Io,
-    SerdeError, Serde,
-    UrlEncodedError, UrlEncoded,
-    UrlError, Url,
-    ApiError, Api,
-    #[cfg(feature = "toml")] TomlSerError, TomlSer,
-    #[cfg(feature = "toml")] TomlDeError, TomlDe,
-    HeaderStrError, HeaderStrError,
-    HeaderParseError, HeaderParseError,
-    #[cfg(feature = "env")] EnvyError, Envy,
-    SerdeQsError, SerdeQs,
-    WebSocketError, WebSocket,
-    String, Other,
+    HttpError => Http,
+    IoError => Io,
+    SerdeError => Serde,
+    UrlEncodedError => UrlEncoded,
+    UrlError => Url,
+    ApiError => Api,
+    #[cfg(feature = "toml")] TomlSerError => TomlSer,
+    #[cfg(feature = "toml")] TomlDeError => TomlDe,
+    HeaderStrError => HeaderStrError,
+    HeaderParseError => HeaderParseError,
+    #[cfg(feature = "env")] EnvyError => Envy,
+    SerdeQsError => SerdeQs,
+    WebSocketError => WebSocket,
+    String => Other,
+    TryFromIntError => IntConversion,
+    WebSocketMessage => UnrecognizedStreamMessage,
 }
 
 #[macro_export]
@@ -157,8 +168,7 @@ from! {
 macro_rules! format_err {
     ( $( $arg:tt )* ) => {
         {
-            use elefren::Error;
-            Error::Other(format!($($arg)*))
+            crate::Error::Other(format!($($arg)*))
         }
     }
 }
@@ -177,9 +187,9 @@ mod tests {
         };
     }
 
-    #[test]
-    fn from_http_error() {
-        let err: HttpError = reqwest::get("not an actual URL").unwrap_err();
+    #[tokio::test]
+    async fn from_http_error() {
+        let err: HttpError = reqwest::get("not an actual URL").await.unwrap_err();
         let err: Error = Error::from(err);
         assert_is!(err, Error::Http(..));
     }
diff --git a/src/event_stream.rs b/src/event_stream.rs
new file mode 100644
index 0000000..c7645e7
--- /dev/null
+++ b/src/event_stream.rs
@@ -0,0 +1,91 @@
+use crate::{
+    entities::{event::Event, prelude::Notification, status::Status},
+    errors::Result,
+    Error,
+};
+use futures::{stream::try_unfold, TryStream};
+use log::debug;
+use tungstenite::Message;
+
+/// Returns a stream of events at the given url location.
+pub fn event_stream(
+    location: impl AsRef<str>,
+) -> Result<impl TryStream<Ok = Event, Error = Error, Item = Result<Event>>> {
+    let (client, response) = tungstenite::connect(location.as_ref())?;
+    let status = response.status();
+    if !status.is_success() {
+        return Err(Error::Api(crate::ApiError {
+            error: status.canonical_reason().map(String::from),
+            error_description: None,
+        }));
+    }
+    Ok(try_unfold(client, |mut client| async move {
+        let mut lines = vec![];
+        loop {
+            match client.read_message() {
+                Ok(Message::Text(message)) => {
+                    let line = message.trim().to_string();
+                    if line.starts_with(":") || line.is_empty() {
+                        continue;
+                    }
+                    lines.push(line);
+                    if let Ok(event) = make_event(&lines) {
+                        lines.clear();
+                        return Ok(Some((event, client)));
+                    } else {
+                        continue;
+                    }
+                },
+                Ok(Message::Ping(data)) => {
+                    debug!("received ping, ponging (metadata: {data:?})");
+                    client.write_message(Message::Pong(data))?;
+                },
+                Ok(message) => return Err(message.into()),
+                Err(err) => return Err(err.into()),
+            }
+        }
+    }))
+}
+
+fn make_event(lines: &[String]) -> Result<Event> {
+    let event;
+    let data;
+    if let Some(event_line) = lines.iter().find(|line| line.starts_with("event:")) {
+        event = event_line[6..].trim().to_string();
+        data = lines
+            .iter()
+            .find(|line| line.starts_with("data:"))
+            .map(|x| x[5..].trim().to_string());
+    } else {
+        #[derive(Deserialize)]
+        struct Message {
+            pub event: String,
+            pub payload: Option<String>,
+        }
+        let message = serde_json::from_str::<Message>(&lines[0])?;
+        event = message.event;
+        data = message.payload;
+    }
+    let event: &str = &event;
+    Ok(match event {
+        "notification" => {
+            let data = data
+                .ok_or_else(|| Error::Other("Missing `data` line for notification".to_string()))?;
+            let notification = serde_json::from_str::<Notification>(&data)?;
+            Event::Notification(notification)
+        },
+        "update" => {
+            let data =
+                data.ok_or_else(|| Error::Other("Missing `data` line for update".to_string()))?;
+            let status = serde_json::from_str::<Status>(&data)?;
+            Event::Update(status)
+        },
+        "delete" => {
+            let data =
+                data.ok_or_else(|| Error::Other("Missing `data` line for delete".to_string()))?;
+            Event::Delete(data)
+        },
+        "filters_changed" => Event::FiltersChanged,
+        _ => return Err(Error::Other(format!("Unknown event `{}`", event))),
+    })
+}
diff --git a/src/helpers/cli.rs b/src/helpers/cli.rs
index 25c1e03..5f4bbf0 100644
--- a/src/helpers/cli.rs
+++ b/src/helpers/cli.rs
@@ -1,10 +1,10 @@
 use std::io::{self, BufRead, Write};
 
-use crate::{errors::Result, http_send::HttpSend, registration::Registered, Mastodon};
+use crate::{errors::Result, registration::Registered, Mastodon};
 
 /// Finishes the authentication process for the given `Registered` object,
 /// using the command-line
-pub fn authenticate<H: HttpSend>(registration: Registered<H>) -> Result<Mastodon<H>> {
+pub async fn authenticate(registration: Registered) -> Result<Mastodon> {
     let url = registration.authorize_url()?;
 
     let stdout = io::stdout();
@@ -20,5 +20,5 @@ pub fn authenticate<H: HttpSend>(registration: Registered<H>) -> Result<Mastodon
     let mut input = String::new();
     stdin.read_line(&mut input)?;
     let code = input.trim();
-    Ok(registration.complete(code)?)
+    registration.complete(code).await
 }
diff --git a/src/http_send.rs b/src/http_send.rs
deleted file mode 100644
index 46ce679..0000000
--- a/src/http_send.rs
+++ /dev/null
@@ -1,26 +0,0 @@
-use crate::Result;
-use reqwest::{Client, Request, RequestBuilder, Response};
-use std::fmt::Debug;
-
-/// Abstracts away the process of turning an HTTP request into an HTTP response
-pub trait HttpSend: Clone + Debug {
-    /// Converts an HTTP request into an HTTP response
-    fn execute(&self, client: &Client, request: Request) -> Result<Response>;
-
-    /// Convenience method so that .build() doesn't have to be called at every
-    /// call site
-    fn send(&self, client: &Client, builder: RequestBuilder) -> Result<Response> {
-        let request = builder.build()?;
-        self.execute(client, request)
-    }
-}
-
-#[doc(hidden)]
-#[derive(Clone, Copy, Debug, PartialEq)]
-pub struct HttpSender;
-
-impl HttpSend for HttpSender {
-    fn execute(&self, client: &Client, request: Request) -> Result<Response> {
-        Ok(client.execute(request)?)
-    }
-}
diff --git a/src/lib.rs b/src/lib.rs
index a2ecfc3..790e06a 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,43 +1,55 @@
-//! // Elefren: API Wrapper around the Mastodon API.
+//! # Elefren: API Wrapper around the Mastodon API.
 //!
 //! Most of the api is documented on [Mastodon's website](https://docs.joinmastodon.org/client/intro/)
 //!
 //! ```no_run
 //! use elefren::{helpers::cli, prelude::*};
+//! use futures_util::StreamExt;
 //!
-//! let registration = Registration::new("https://botsin.space")
-//!     .client_name("elefren_test")
-//!     .build()
-//!     .unwrap();
-//! let mastodon = cli::authenticate(registration).unwrap();
+//! tokio_test::block_on(async {
+//!     let registration = Registration::new("https://botsin.space")
+//!         .client_name("elefren_test")
+//!         .build()
+//!         .await
+//!         .unwrap();
+//!     let mastodon = cli::authenticate(registration).await.unwrap();
 //!
-//! println!(
-//!     "{:?}",
-//!     mastodon
-//!         .get_home_timeline()
-//!         .unwrap()
-//!         .items_iter()
-//!         .take(100)
-//!         .collect::<Vec<_>>()
-//! );
+//!     println!(
+//!         "{:?}",
+//!         mastodon
+//!             .get_home_timeline()
+//!             .await
+//!             .unwrap()
+//!             .items_iter()
+//!             .take(100)
+//!             .collect::<Vec<_>>()
+//!             .await
+//!     );
+//! });
 //! ```
 //!
 //! Elefren also supports Mastodon's Streaming API:
 //!
-//! // Example
+//! ## Example
 //!
 //! ```no_run
 //! use elefren::{prelude::*, entities::event::Event};
+//! use futures_util::TryStreamExt;
+//!
 //! let data = Data::default();
 //! let client = Mastodon::from(data);
-//! for event in client.streaming_user().unwrap() {
-//!     match event {
-//!         Event::Update(ref status) => { /* .. */ },
-//!         Event::Notification(ref notification) => { /* .. */ },
-//!         Event::Delete(ref id) => { /* .. */ },
-//!         Event::FiltersChanged => { /* .. */ },
-//!     }
-//! }
+//! tokio_test::block_on(async {
+//!     let stream = client.streaming_user().await.unwrap();
+//!     stream.try_for_each(|event| async move {
+//!         match event {
+//!             Event::Update(ref status) => { /* .. */ },
+//!             Event::Notification(ref notification) => { /* .. */ },
+//!             Event::Delete(ref id) => { /* .. */ },
+//!             Event::FiltersChanged => { /* .. */ },
+//!         }
+//!         Ok(())
+//!     }).await.unwrap();
+//! });
 //! ```
 
 #![deny(
@@ -53,7 +65,7 @@
     unused_qualifications
 )]
 
-#[macro_use]
+// #[macro_use]
 extern crate log;
 #[macro_use]
 extern crate doc_comment;
@@ -84,20 +96,13 @@ extern crate tempfile;
 #[cfg_attr(all(test, any(feature = "toml", feature = "json")), macro_use)]
 extern crate indoc;
 
-use std::{borrow::Cow, io::BufRead, ops};
-
-use reqwest::{Client, RequestBuilder, Response};
-use tap_reader::Tap;
-use tungstenite::client::AutoStream;
-
-use entities::prelude::*;
-use http_send::{HttpSend, HttpSender};
 use page::Page;
 
 pub use data::Data;
 pub use errors::{ApiError, Error, Result};
 pub use isolang::Language;
-pub use mastodon_client::{MastodonClient, MastodonUnauthenticated};
+pub use mastodon::{Mastodon, MastodonUnauthenticated};
+// pub use mastodon_client::{MastodonClient, MastodonUnauthenticated};
 pub use registration::Registration;
 pub use requests::{
     AddFilterRequest,
@@ -116,11 +121,10 @@ pub mod data;
 pub mod entities;
 /// Errors
 pub mod errors;
+/// Event stream generators
+pub mod event_stream;
 /// Collection of helpers for serializing/deserializing `Data` objects
 pub mod helpers;
-/// Contains trait for converting `reqwest::Request`s to `reqwest::Response`s
-pub mod http_send;
-mod mastodon_client;
 /// Handling multiple pages of entities.
 pub mod page;
 /// Registering your app.
@@ -139,670 +143,12 @@ pub mod prelude {
         scopes::Scopes,
         Data,
         Mastodon,
-        MastodonClient,
+        // MastodonClient,
         NewStatus,
         Registration,
         StatusBuilder,
         StatusesRequest,
     };
 }
-
-/// Your mastodon application client, handles all requests to and from Mastodon.
-#[derive(Clone, Debug)]
-pub struct Mastodon<H: HttpSend = HttpSender> {
-    client: Client,
-    http_sender: H,
-    /// Raw data about your mastodon instance.
-    pub data: Data,
-}
-
-impl<H: HttpSend> Mastodon<H> {
-    methods![get, post, delete,];
-
-    fn route(&self, url: &str) -> String {
-        format!("{}{}", self.base, url)
-    }
-
-    pub(crate) fn send(&self, req: RequestBuilder) -> Result<Response> {
-        Ok(self
-            .http_sender
-            .send(&self.client, req.bearer_auth(&self.token))?)
-    }
-}
-
-impl From<Data> for Mastodon<HttpSender> {
-    /// Creates a mastodon instance from the data struct.
-    fn from(data: Data) -> Mastodon<HttpSender> {
-        let mut builder = MastodonBuilder::new(HttpSender);
-        builder.data(data);
-        builder
-            .build()
-            .expect("We know `data` is present, so this should be fine")
-    }
-}
-
-impl<H: HttpSend> MastodonClient<H> for Mastodon<H> {
-    type Stream = EventReader<WebSocket>;
-
-    paged_routes! {
-        (get) favourites: "favourites" => Status,
-        (get) blocks: "blocks" => Account,
-        (get) domain_blocks: "domain_blocks" => String,
-        (get) follow_requests: "follow_requests" => Account,
-        (get) get_home_timeline: "timelines/home" => Status,
-        (get) get_emojis: "custom_emojis" => Emoji,
-        (get) mutes: "mutes" => Account,
-        (get) notifications: "notifications" => Notification,
-        (get) reports: "reports" => Report,
-        (get (q: &'a str, #[serde(skip_serializing_if = "Option::is_none")] limit: Option<u64>, following: bool,)) search_accounts: "accounts/search" => Account,
-        (get) get_endorsements: "endorsements" => Account,
-    }
-
-    paged_routes_with_id! {
-        (get) followers: "accounts/{}/followers" => Account,
-        (get) following: "accounts/{}/following" => Account,
-        (get) reblogged_by: "statuses/{}/reblogged_by" => Account,
-        (get) favourited_by: "statuses/{}/favourited_by" => Account,
-    }
-
-    route! {
-        (delete (domain: String,)) unblock_domain: "domain_blocks" => Empty,
-        (get) instance: "instance" => Instance,
-        (get) verify_credentials: "accounts/verify_credentials" => Account,
-