cel
/
mastodon-async
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add logging

This commit is contained in:
D. Scott Boggs 2022-12-07 15:58:28 -05:00
parent fdf180398f
commit c9fc25a0c9
24 changed files with 515 additions and 168 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "elefren"
version = "0.23.0"
version = "0.24.0"
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"
@ -16,7 +16,6 @@ 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"
@ -25,6 +24,15 @@ tungstenite = "0.18"
url = "1"
# Provides parsing for the link header in get_links() in page.rs
hyper-old-types = "0.11.0"
futures-util = "0.3.25"
[dependencies.uuid]
version = "1.2.2"
features = ["v4"]
[dependencies.log]
version = "0.4"
features = ["kv_unstable", "serde", "std", "kv_unstable_serde", "kv_unstable_std"]
[dependencies.chrono]
version = "0.4"
@ -40,7 +48,7 @@ features = ["serde"]
[dependencies.reqwest]
version = "0.11"
features = ["multipart", "json"]
features = ["multipart", "json", "stream"]
[dependencies.serde]
version = "1"
@ -50,6 +58,11 @@ features = ["derive"]
version = "0.5"
optional = true
[dependencies.tokio]
version = "1.22.0"
features = ["rt-multi-thread", "macros"]
[dev-dependencies]
tokio-test = "0.4.2"
futures-util = "0.3.25"
@ -61,10 +74,6 @@ 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"]

4
src/entities/account.rs
View File

@ -10,7 +10,7 @@ use serde::{
use std::path::PathBuf;
/// A struct representing an Account.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
pub struct Account {
/// Equals `username` for local users, includes `@domain` for remote ones.
pub acct: String,
@ -74,7 +74,7 @@ impl MetadataField {
}
/// An extra object given from `verify_credentials` giving defaults about a user
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
pub struct Source {
privacy: Option<status_builder::Visibility>,
#[serde(deserialize_with = "string_or_bool")]

10
src/entities/attachment.rs
View File

@ -1,9 +1,9 @@
//! Module containing everything related to media attachements.
use serde::Deserialize;
use serde::{Deserialize, Serialize};
/// A struct representing a media attachment.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
pub struct Attachment {
/// ID of the attachment.
pub id: String,
@ -26,7 +26,7 @@ pub struct Attachment {
}
/// Information about the attachment itself.
#[derive(Debug, Deserialize, Clone, PartialEq)]
#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)]
pub struct Meta {
/// Original version.
pub original: Option<ImageDetails>,
@ -35,7 +35,7 @@ pub struct Meta {
}
/// Dimensions of an attachement.
#[derive(Debug, Deserialize, Clone, PartialEq)]
#[derive(Debug, Deserialize, Serialize, Clone, PartialEq)]
pub struct ImageDetails {
/// width of attachment.
width: u64,
@ -48,7 +48,7 @@ pub struct ImageDetails {
}
/// The type of media attachment.
#[derive(Debug, Deserialize, Clone, Copy, PartialEq)]
#[derive(Debug, Deserialize, Serialize, Clone, Copy, PartialEq)]
pub enum MediaType {
/// An image.
#[serde(rename = "image")]

4
src/entities/card.rs
View File

@ -1,9 +1,9 @@
//! Module representing cards of statuses.
use serde::Deserialize;
use serde::{Deserialize, Serialize};
/// A card of a status.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
pub struct Card {
/// The url associated with the card.
pub url: String,

4
src/entities/context.rs
View File

@ -1,12 +1,12 @@
//! A module about contexts of statuses.
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use super::status::Status;
/// A context of a status returning a list of statuses it replied to and
/// statuses replied to it.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Context {
/// Statuses that were replied to.
pub ancestors: Vec<Status>,

3
src/entities/event.rs
View File

@ -1,6 +1,7 @@
use crate::entities::{notification::Notification, status::Status};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone)]
#[derive(Debug, Clone, Deserialize, Serialize)]
/// Events that come from the /streaming/user API call
pub enum Event {
/// Update event

8
src/entities/instance.rs
View File

@ -1,10 +1,10 @@
//! Module containing everything related to an instance.
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use super::account::Account;
/// A struct containing info of an instance.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
pub struct Instance {
/// URI of the current instance
pub uri: String,
@ -32,14 +32,14 @@ pub struct Instance {
}
/// Object containing url for streaming api.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
pub struct StreamingApi {
/// Url for streaming API, typically a `wss://` url.
pub streaming_api: String,
}
/// Statistics about the Mastodon instance.
#[derive(Debug, Clone, Copy, Deserialize, PartialEq)]
#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq)]
pub struct Stats {
user_count: u64,
status_count: u64,

57
src/entities/itemsiter.rs
View File

@ -1,7 +1,8 @@
use futures::{stream::unfold, Stream};
use log::{as_debug, as_serde, debug, info, warn};
use crate::page::Page;
use serde::Deserialize;
use serde::{Deserialize, Serialize};
/// Abstracts away the `next_page` logic into a single stream of items
///
@ -22,14 +23,14 @@ use serde::Deserialize;
///
/// See documentation for `futures::Stream::StreamExt` for available methods.
#[derive(Debug, Clone)]
pub(crate) struct ItemsIter<T: Clone + for<'de> Deserialize<'de>> {
pub(crate) struct ItemsIter<T: Clone + for<'de> Deserialize<'de> + Serialize> {
page: Page<T>,
buffer: Vec<T>,
cur_idx: usize,
use_initial: bool,
}
impl<'a, T: Clone + for<'de> Deserialize<'de>> ItemsIter<T> {
impl<'a, T: Clone + for<'de> Deserialize<'de> + Serialize> ItemsIter<T> {
pub(crate) fn new(page: Page<T>) -> ItemsIter<T> {
ItemsIter {
page,
@ -40,43 +41,50 @@ impl<'a, T: Clone + for<'de> Deserialize<'de>> ItemsIter<T> {
}
fn need_next_page(&self) -> bool {
self.buffer.is_empty() || self.cur_idx == self.buffer.len()
if self.buffer.is_empty() || self.cur_idx == self.buffer.len() {
debug!(idx = self.cur_idx, buffer_len = self.buffer.len(); "next page needed");
true
} else {
false
}
}
async fn fill_next_page(&mut self) -> Option<()> {
let items = if let Ok(items) = self.page.next_page().await {
items
} else {
return None;
};
if let Some(items) = items {
if items.is_empty() {
return None;
}
self.buffer = items;
self.cur_idx = 0;
Some(())
} else {
None
match self.page.next_page().await {
Ok(Some(items)) => {
info!(item_count = items.len(); "next page received");
if items.is_empty() {
return None;
}
self.buffer = items;
self.cur_idx = 0;
Some(())
},
Err(err) => {
warn!(err = as_debug!(err); "error encountered filling next page");
None
},
_ => 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()
{
let idx = this.cur_idx;
if this.page.initial_items.is_empty() || idx == this.page.initial_items.len() {
debug!(index = idx, n_initial_items = this.page.initial_items.len(); "exhausted initial items and no more pages are present");
return None;
}
let idx = this.cur_idx;
if this.cur_idx == this.page.initial_items.len() - 1 {
if idx == this.page.initial_items.len() - 1 {
this.cur_idx = 0;
this.use_initial = false;
debug!(index = idx, n_initial_items = this.page.initial_items.len(); "exhausted initial items");
} else {
this.cur_idx += 1;
}
let item = this.page.initial_items[idx].clone();
debug!(item = as_serde!(item), index = idx; "yielding item from initial items");
// let item = Box::pin(item);
// pin_mut!(item);
Some((item, this))
@ -89,8 +97,7 @@ impl<'a, T: Clone + for<'de> Deserialize<'de>> ItemsIter<T> {
let idx = this.cur_idx;
this.cur_idx += 1;
let item = this.buffer[idx].clone();
// let item = Box::pin(item);
// pin_mut!(item);
debug!(item = as_serde!(item), index = idx; "yielding item from initial stream");
Some((item, this))
}
})

4
src/entities/list.rs
View File

@ -1,7 +1,7 @@
use serde::Deserialize;
use serde::{Deserialize, Serialize};
/// Used for ser/de of list resources
#[derive(Clone, Debug, Deserialize, PartialEq)]
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
pub struct List {
id: String,
title: String,

4
src/entities/mention.rs
View File

@ -1,5 +1,7 @@
use serde::{Deserialize, Serialize};
/// Represents a `mention` used in a status
#[derive(Debug, Clone, PartialEq)]
#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
pub struct Mention {
/// URL of user's profile (can be remote)
pub url: String,

2
src/entities/mod.rs
View File

@ -33,7 +33,7 @@ pub mod search_result;
pub mod status;
/// An empty JSON object.
#[derive(Deserialize, Debug, Copy, Clone, PartialEq)]
#[derive(Deserialize, Serialize, Debug, Copy, Clone, PartialEq)]
pub struct Empty {}
/// The purpose of this module is to alleviate imports of many common

6
src/entities/notification.rs
View File

@ -2,10 +2,10 @@
use super::{account::Account, status::Status};
use chrono::prelude::*;
use serde::Deserialize;
use serde::{Deserialize, Serialize};
/// A struct containing info about a notification.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
pub struct Notification {
/// The notification ID.
pub id: String,
@ -21,7 +21,7 @@ pub struct Notification {
}
/// The type of notification.
#[derive(Debug, Clone, Copy, Deserialize, PartialEq)]
#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq)]
#[serde(rename_all = "lowercase")]
pub enum NotificationType {
/// Someone mentioned the application client in another status.

2
src/entities/push.rs
View File

@ -14,7 +14,7 @@ pub struct Alerts {
}
/// Represents a new Push subscription
#[derive(Debug, Clone, PartialEq, Deserialize)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Subscription {
/// The `id` of the subscription
pub id: String,

4
src/entities/relationship.rs
View File

@ -1,10 +1,10 @@
//! module containing everything relating to a relationship with
//! another account.
use serde::Deserialize;
use serde::{Deserialize, Serialize};
/// A struct containing information about a relationship with another account.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Relationship {
/// Target account id
pub id: String,

4
src/entities/report.rs
View File

@ -1,9 +1,9 @@
//! module containing information about a finished report of a user.
use serde::Deserialize;
use serde::{Deserialize, Serialize};
/// A struct containing info about a report.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Report {
/// The ID of the report.
pub id: String,

6
src/entities/search_result.rs
View File

@ -1,6 +1,6 @@
//! A module containing info relating to a search result.
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use super::{
prelude::{Account, Status},
@ -8,7 +8,7 @@ use super::{
};
/// A struct containing results of a search.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SearchResult {
/// An array of matched Accounts.
pub accounts: Vec<Account>,
@ -20,7 +20,7 @@ pub struct SearchResult {
/// A struct containing results of a search, with `Tag` objects in the
/// `hashtags` field
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct SearchResultV2 {
/// An array of matched Accounts.
pub accounts: Vec<Account>,

12
src/entities/status.rs
View File

@ -3,10 +3,10 @@
use super::prelude::*;
use crate::{entities::card::Card, status_builder::Visibility};
use chrono::prelude::*;
use serde::Deserialize;
use serde::{Deserialize, Serialize};
/// A status from the instance.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Status {
/// The ID of the status.
pub id: String,
@ -65,7 +65,7 @@ pub struct Status {
}
/// A mention of another user.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Mention {
/// URL of user's profile (can be remote).
pub url: String,
@ -78,7 +78,7 @@ pub struct Mention {
}
/// Struct representing an emoji within text.
#[derive(Clone, Debug, Deserialize, PartialEq)]
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
pub struct Emoji {
/// The shortcode of the emoji
pub shortcode: String,
@ -89,7 +89,7 @@ pub struct Emoji {
}
/// Hashtags in the status.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Tag {
/// The hashtag, not including the preceding `#`.
pub name: String,
@ -98,7 +98,7 @@ pub struct Tag {
}
/// Application details.
#[derive(Debug, Clone, Deserialize, PartialEq)]
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct Application {
/// Name of the application.
pub name: String,

28
src/event_stream.rs
View File

@ -4,40 +4,51 @@ use crate::{
Error,
};
use futures::{stream::try_unfold, TryStream};
use log::debug;
use log::{as_debug, as_serde, debug, error, info, trace};
use tungstenite::Message;
/// Returns a stream of events at the given url location.
pub fn event_stream(
location: impl AsRef<str>,
location: String,
) -> Result<impl TryStream<Ok = Event, Error = Error, Item = Result<Event>>> {
let (client, response) = tungstenite::connect(location.as_ref())?;
trace!(location = location; "connecting to websocket for events");
let (client, response) = tungstenite::connect(&location)?;
let status = response.status();
if !status.is_success() {
error!(
status = as_debug!(status),
body = response.body().as_ref().map(|it| String::from_utf8_lossy(it.as_slice())).unwrap_or("(empty body)".into()),
location = &location;
"error connecting to websocket"
);
return Err(Error::Api(crate::ApiError {
error: status.canonical_reason().map(String::from),
error_description: None,
}));
}
Ok(try_unfold(client, |mut client| async move {
debug!(location = &location, status = as_debug!(status); "successfully connected to websocket");
Ok(try_unfold((client, location), |mut this| async move {
let (ref mut client, ref location) = this;
let mut lines = vec![];
loop {
match client.read_message() {
Ok(Message::Text(message)) => {
let line = message.trim().to_string();
Ok(Message::Text(line)) => {
debug!(message = line, location = &location; "received websocket message");
let line = line.trim().to_string();
if line.starts_with(":") || line.is_empty() {
continue;
}
lines.push(line);
if let Ok(event) = make_event(&lines) {
info!(event = as_serde!(event), location = location; "received websocket event");
lines.clear();
return Ok(Some((event, client)));
return Ok(Some((event, this)));
} else {
continue;
}
},
Ok(Message::Ping(data)) => {
debug!("received ping, ponging (metadata: {data:?})");
debug!(metadata = as_serde!(data); "received ping, ponging");
client.write_message(Message::Pong(data))?;
},
Ok(message) => return Err(message.into()),
@ -67,6 +78,7 @@ fn make_event(lines: &[String]) -> Result<Event> {
data = message.payload;
}
let event: &str = &event;
trace!(event = event, payload = data; "websocket message parsed");
Ok(match event {
"notification" => {
let data = data

70
src/helpers/log.rs Normal file
View File

@ -0,0 +1,70 @@
use serde::Serialize;
/// Log metadata about this request based on the type given:
///
/// ```no_run
/// use elefren::log_serde;
/// tokio_test::block_on(async {
/// let request = reqwest::get("https://example.org/").await.unwrap();
/// log::warn!(
/// status = log_serde!(request Status),
/// headers = log_serde!(request Headers);
/// "test"
/// );
/// })
/// ```
#[macro_export]
macro_rules! log_serde {
($response:ident $type_name:tt) => {
log::as_serde!($crate::helpers::log::$type_name::from(&$response))
};
}
/// Serializable form of reqwest's Status type.
#[derive(Debug, Copy, Clone, Serialize, Deserialize)]
pub struct Status {
/// The numerical representation of the status
pub code: u16,
/// it's canonical reason.
pub message: Option<&'static str>,
}
impl Status {
/// New from reqwest's Status type (which is more useful but not
/// serializable).
pub fn new(status: reqwest::StatusCode) -> Self {
Self {
code: status.as_u16(),
message: status.canonical_reason(),
}
}
}
impl From<&reqwest::Response> for Status {
fn from(value: &reqwest::Response) -> Self {
Self::new(value.status())
}
}
/// Helper for logging request headers
#[derive(Debug)]
pub struct Headers(pub reqwest::header::HeaderMap);
impl Serialize for Headers {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
serializer.collect_map(
self.0
.iter()
.map(|(k, v)| (format!("{k:?}"), format!("{v:?}"))),
)
}
}
impl From<&reqwest::Response> for Headers {
fn from(value: &reqwest::Response) -> Self {
Headers(value.headers().clone())
}
}

5
src/helpers/mod.rs
View File

@ -36,3 +36,8 @@ pub mod env;
/// Helpers for working with the command line
pub mod cli;
/// Helpers for serializing data for logging
pub mod log;
/// Adapter for reading JSON data from a response with better logging and a
/// fail-safe timeout.
pub mod read_response;

196
src/macros.rs
View File

@ -1,18 +1,45 @@
macro_rules! methods {
($($method:ident,)+) => {
($($method:ident and $method_with_call_id:ident,)+) => {
$(
async fn $method<T: for<'de> serde::Deserialize<'de>>(&self, url: impl AsRef<str>) -> Result<T>
{
let url = url.as_ref();
Ok(
self.client
doc_comment! {
concat!("Make a ", stringify!($method), " API request, and deserialize the result into T"),
async fn $method<T: for<'de> serde::Deserialize<'de> + serde::Serialize>(&self, url: impl AsRef<str>) -> Result<T>
{
let call_id = uuid::Uuid::new_v4();
self.$method_with_call_id(url, call_id).await
}
}
doc_comment! {
concat!(
"Make a ", stringify!($method), " API request, and deserialize the result into T.\n\n",
"Logging will use the provided UUID, rather than generating one before making the request.",
),
async fn $method_with_call_id<T: for<'de> serde::Deserialize<'de> + serde::Serialize>(&self, url: impl AsRef<str>, call_id: Uuid) -> Result<T>
{
use log::{debug, error, as_debug, as_serde};
let url = url.as_ref();
debug!(url = url, method = stringify!($method), call_id = as_debug!(call_id); "making API request");
let response = self.client
.$method(url)
.send()
.await?
.error_for_status()?
.json()
.await?
)
.await?;
match response.error_for_status() {
Ok(response) => {
let response = response
.json()
.await?;
debug!(response = as_serde!(response), url = url, method = stringify!($method), call_id = as_debug!(call_id); "received API response");
Ok(response)
}
Err(err) => {
error!(err = as_debug!(err), url = url, method = stringify!($method), call_id = as_debug!(call_id); "error making API request");
Err(err.into())
}
}
}
}
)+
};
@ -35,10 +62,21 @@ macro_rules! paged_routes {
"```"
),
pub async fn $name(&self) -> Result<Page<$ret>> {
use log::{debug, as_debug, error};
let url = self.route(concat!("/api/v1/", $url));
let call_id = uuid::Uuid::new_v4();
debug!(url = url, method = stringify!($method), call_id = as_debug!(call_id); "making API request");
let response = self.client.$method(&url).send().await?;
Page::new(self.clone(), response).await
match response.error_for_status() {
Ok(response) => {
Page::new(self.clone(), response, call_id).await
}
Err(err) => {
error!(err = as_debug!(err), url = url, method = stringify!($method), call_id = as_debug!(call_id); "error making API request");
Err(err.into())
}
}
}
}
@ -55,6 +93,9 @@ macro_rules! paged_routes {
),
pub async fn $name<'a>(&self, $($param: $typ,)*) -> Result<Page<$ret>> {
use serde_urlencoded;
use log::{debug, as_debug, error};
let call_id = uuid::Uuid::new_v4();
#[derive(Serialize)]
struct Data<'a> {
@ -79,9 +120,19 @@ macro_rules! paged_routes {
let url = format!(concat!("/api/v1/", $url, "?{}"), &qs);
debug!(url = url, method = "get", call_id = as_debug!(call_id); "making API request");
let response = self.client.get(&url).send().await?;
Page::new(self.clone(), response).await
match response.error_for_status() {
Ok(response) => {
Page::new(self.clone(), response, call_id).await
}
Err(err) => {
error!(err = as_debug!(err), url = url, method = stringify!($method), call_id = as_debug!(call_id); "error making API request");
Err(err.into())
}
}
}
}
@ -101,6 +152,10 @@ macro_rules! route_v2 {
),
pub async fn $name<'a>(&self, $($param: $typ,)*) -> Result<$ret> {
use serde_urlencoded;
use log::{debug, as_serde};
use uuid::Uuid;
let call_id = Uuid::new_v4();
#[derive(Serialize)]
struct Data<'a> {
@ -120,9 +175,11 @@ macro_rules! route_v2 {
let qs = serde_urlencoded::to_string(&qs_data)?;
debug!(query_string_data = as_serde!(qs_data); "URL-encoded data to be sent in API request");
let url = format!(concat!("/api/v2/", $url, "?{}"), &qs);
self.get(self.route(&url)).await
self.get_with_call_id(self.route(&url), call_id).await
}
}
@ -143,36 +200,57 @@ macro_rules! route {
pub async fn $name(&self, $($param: $typ,)*) -> Result<$ret> {
use reqwest::multipart::{Form, Part};
use std::io::Read;
use log::{debug, error, as_debug, as_serde};
use uuid::Uuid;
let call_id = Uuid::new_v4();
let form_data = Form::new()
$(
.part(stringify!($param), {
let mut file = std::fs::File::open($param.as_ref())?;
let mut data = if let Ok(metadata) = file.metadata() {
Vec::with_capacity(metadata.len().try_into()?)
} else {
vec![]
};
file.read_to_end(&mut data)?;
Part::bytes(data)
match std::fs::File::open($param.as_ref()) {
Ok(mut file) => {
let mut data = if let Ok(metadata) = file.metadata() {
Vec::with_capacity(metadata.len().try_into()?)
} else {
vec![]
};
file.read_to_end(&mut data)?;
Part::bytes(data)
}
Err(err) => {
error!(path = $param.as_ref(), error = as_debug!(err); "error reading file contents for multipart form");
return Err(err.into());
}
}
})
)*;
let url = &self.route(concat!("/api/v1/", $url));
debug!(
url = url, method = stringify!($method),
multipart_form_data = as_debug!(form_data), call_id = as_debug!(call_id);
"making API request"
);
let response = self.client
.post(&self.route(concat!("/api/v1/", $url)))
.post(url)
.multipart(form_data)
.send()
.await?;
let status = response.status().clone();
if status.is_client_error() {
return Err(Error::Client(status));
} else if status.is_server_error() {
return Err(Error::Server(status));
match response.error_for_status() {
Ok(response) => {
let response = response.json().await?;
debug!(response = as_serde!(response), url = url, method = stringify!($method), call_id = as_debug!(call_id); "received API response");
Ok(response)
}
Err(err) => {
error!(err = as_debug!(err), url = url, method = stringify!($method), call_id = as_debug!(call_id); "error making API request");
Err(err.into())
}
}
Ok(response.json().await?)
}
}
@ -188,6 +266,10 @@ macro_rules! route {
),
pub async fn $name<'a>(&self, $($param: $typ,)*) -> Result<$ret> {
use serde_urlencoded;
use log::{debug, as_serde};
use uuid::Uuid;
let call_id = Uuid::new_v4();
#[derive(Serialize)]
struct Data<'a> {
@ -205,11 +287,14 @@ macro_rules! route {
_marker: ::std::marker::PhantomData,
};
let qs = serde_urlencoded::to_string(&qs_data)?;
debug!(query_string_data = as_serde!(qs_data); "URL-encoded data to be sent in API request");
let url = format!(concat!("/api/v1/", $url, "?{}"), &qs);
self.get(self.route(&url)).await
self.get_with_call_id(self.route(&url), call_id).await
}
}
@ -224,12 +309,22 @@ macro_rules! route {
"`\n# Errors\nIf `access_token` is not set.",
),
pub async fn $name(&self, $($param: $typ,)*) -> Result<$ret> {
use log::{debug, error, as_debug, as_serde};
use uuid::Uuid;
let call_id = Uuid::new_v4();
let form_data = json!({
$(
stringify!($param): $param,
)*
});
debug!(
url = $url, method = stringify!($method),
call_id = as_debug!(call_id),
form_data = as_serde!(&form_data);
"making API request"
);
let response = self.client
.$method(&self.route(concat!("/api/v1/", $url)))
@ -237,15 +332,17 @@ macro_rules! route {
.send()
.await?;
let status = response.status().clone();
if status.is_client_error() {
return Err(Error::Client(status));
} else if status.is_server_error() {
return Err(Error::Server(status));
match response.error_for_status() {
Ok(response) => {
let response = response.json().await?;
debug!(response = as_serde!(response), url = $url, method = stringify!($method), call_id = as_debug!(call_id); "received API response");
Ok(response)
}
Err(err) => {
error!(err = as_debug!(err), url = $url, method = stringify!($method), call_id = as_debug!(call_id); "error making API request");
Err(err.into())
}
}
Ok(response.json().await?)
}
}
@ -321,10 +418,23 @@ macro_rules! paged_routes_with_id {
"```"
),
pub async fn $name(&self, id: &str) -> Result<Page<$ret>> {
let url = self.route(&format!(concat!("/api/v1/", $url), id));
let response = self.client.$method(&url).send().await?;
use log::{debug, error, as_debug};
use uuid::Uuid;
Page::new(self.clone(), response).await
let call_id = Uuid::new_v4();
let url = self.route(&format!(concat!("/api/v1/", $url), id));
debug!(url = url, method = stringify!($method), call_id = as_debug!(call_id); "making API request");
let response = self.client.$method(&url).send().await?;
match response.error_for_status() {
Ok(response) => {
Page::new(self.clone(), response, call_id).await
}
Err(err) => {
error!(err = as_debug!(err), url = url, method = stringify!($method), call_id = as_debug!(call_id); "error making API request");
Err(err.into())
}
}
}
}

109
src/mastodon.rs
View File

@ -20,8 +20,10 @@ use crate::{
UpdatePushRequest,
};
use futures::TryStream;
use log::{as_debug, as_serde, debug, error, info, trace};
use reqwest::Client;
use url::Url;
use uuid::Uuid;
/// The Mastodon client is a smart pointer to this struct
#[derive(Clone, Debug)]
@ -54,7 +56,7 @@ impl From<Data> for Mastodon {
}
}
impl Mastodon {
methods![get, post, delete,];
methods![get and get_with_call_id, post and post_with_call_id, delete and delete_with_call_id,];
paged_routes! {
(get) favourites: "favourites" => Status,
@ -235,20 +237,29 @@ impl Mastodon {
where
S: Into<Option<StatusesRequest<'a>>>,
{
let call_id = Uuid::new_v4();
let mut url = format!("{}/api/v1/accounts/{}/statuses", self.data.base, id);
if let Some(request) = request.into() {
url = format!("{}{}", url, request.to_querystring()?);
}
debug!(url = url, method = stringify!($method), call_id = as_debug!(call_id); "making API request");
let response = self.client.get(&url).send().await?;
Page::new(self.clone(), response).await
match response.error_for_status() {
Ok(response) => Page::new(self.clone(), response, call_id).await,
Err(err) => {
error!(err = as_debug!(err), url = url, method = stringify!($method), call_id = as_debug!(call_id); "error making API request");
Err(err.into())
},
}
}
/// Returns the client account's relationship to a list of other accounts.
/// Such as whether they follow them or vice versa.
pub async fn relationships(&self, ids: &[&str]) -> Result<Page<Relationship>> {
let call_id = Uuid::new_v4();
let mut url = self.route("/api/v1/accounts/relationships?");
if ids.len() == 1 {
@ -263,36 +274,78 @@ impl Mastodon {
url.pop();
}
debug!(
url = url, method = stringify!($method),
call_id = as_debug!(call_id), account_ids = as_serde!(ids);
"making API request"
);
let response = self.client.get(&url).send().await?;
Page::new(self.clone(), response).await
match response.error_for_status() {
Ok(response) => Page::new(self.clone(), response, call_id).await,
Err(err) => {
error!(
err = as_debug!(err), url = url,
method = stringify!($method), call_id = as_debug!(call_id),
account_ids = as_serde!(ids);
"error making API request"
);
Err(err.into())
},
}
}
/// Add a push notifications subscription
pub async fn add_push_subscription(&self, request: &AddPushRequest) -> Result<Subscription> {
let call_id = Uuid::new_v4();
let request = request.build()?;
Ok(self
.client
.post(&self.route("/api/v1/push/subscription"))
.json(&request)
.send()
.await?
.json()
.await?)
let url = &self.route("/api/v1/push/subscription");
debug!(
url = url, method = stringify!($method),
call_id = as_debug!(call_id), post_body = as_serde!(request);
"making API request"
);
let response = self.client.post(url).json(&request).send().await?;
match response.error_for_status() {
Ok(response) => {
let status = response.status();
let response = response.json().await?;
debug!(status = as_debug!(status), response = as_serde!(response); "received API response");
Ok(response)
},
Err(err) => {
error!(err = as_debug!(err), url = url, method = stringify!($method), call_id = as_debug!(call_id); "error making API request");
Err(err.into())
},
}
}
/// Update the `data` portion of the push subscription associated with this
/// access token
pub async fn update_push_data(&self, request: &UpdatePushRequest) -> Result<Subscription> {
let call_id = Uuid::new_v4();
let request = request.build();
Ok(self
.client
.put(&self.route("/api/v1/push/subscription"))
.json(&request)
.send()
.await?
.json()
.await?)
let url = &self.route("/api/v1/push/subscription");
debug!(
url = url, method = stringify!($method),
call_id = as_debug!(call_id), post_body = as_serde!(request);
"making API request"
);
let response = self.client.post(url).json(&request).send().await?;
match response.error_for_status() {
Ok(response) => {
let status = response.status();
let response = response.json().await?;
debug!(status = as_debug!(status), response = as_serde!(response); "received API response");
Ok(response)
},
Err(err) => {
error!(err = as_debug!(err), url = url, method = stringify!($method), call_id = as_debug!(call_id); "error making API request");
Err(err.into())
},
}
}
/// Get all accounts that follow the authenticated user
@ -333,11 +386,22 @@ impl Mastodon {
/// });
/// ```
pub async fn streaming_user(&self) -> Result<impl TryStream<Ok = Event, Error = Error>> {
let call_id = Uuid::new_v4();
let mut url: Url = self.route("/api/v1/streaming").parse()?;
url.query_pairs_mut()
.append_pair("access_token", &self.data.token)
.append_pair("stream", "user");
let mut url: Url = reqwest::get(url.as_str()).await?.url().as_str().parse()?;
debug!(
url = as_debug!(url), call_id = as_debug!(call_id);
"making user streaming API request"
);
let response = reqwest::get(url.as_str()).await?;
let mut url: Url = response.url().as_str().parse()?;
info!(
url = as_debug!(url), call_id = as_debug!(call_id),
status = response.status().as_str();
"received url from streaming API request"
);
let new_scheme = match url.scheme() {
"http" => "ws",
"https" => "wss",
@ -346,12 +410,12 @@ impl Mastodon {
url.set_scheme(new_scheme)
.map_err(|_| Error::Other("Bad URL scheme!".to_string()))?;
event_stream(url)
event_stream(url.to_string())
}
}
impl MastodonUnauthenticated {
methods![get,];
methods![get and get_with_call_id,];
/// Create a new client for unauthenticated requests to a given Mastodon
/// instance.
@ -362,6 +426,7 @@ impl MastodonUnauthenticated {
} else {
format!("https://{}", base.trim_start_matches("http://"))
};
trace!(base = base; "creating new mastodon client");
Ok(MastodonUnauthenticated {
client: Client::new(),
base: Url::parse(&base)?,

67
src/page.rs
View File

@ -2,8 +2,10 @@ use super::{Mastodon, Result};
use crate::{entities::itemsiter::ItemsIter, format_err};
use futures::Stream;
use hyper_old_types::header::{parsing, Link, RelationType};
use log::{as_debug, as_serde, debug, error, trace};
use reqwest::{header::LINK, Response, Url};
use serde::Deserialize;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
// use url::Url;
macro_rules! pages {
@ -18,13 +20,41 @@ macro_rules! pages {
None => return Ok(None),
};
let response = self.mastodon.client.get(url).send().await?;
debug!(
url = as_debug!(url), method = "get",
call_id = as_debug!(self.call_id),
direction = stringify!($direction);
"making API request"
);
let url: String = url.into(); // <- for logging
let response = self.mastodon.client.get(&url).send().await?;
match response.error_for_status() {
Ok(response) => {
let (prev, next) = get_links(&response, self.call_id)?;
let response = response.json().await?;
debug!(
url = url, method = "get", next = as_debug!(next),
prev = as_debug!(prev), call_id = as_debug!(self.call_id),
response = as_serde!(response);
"received next pages from API"
);
self.next = next;
self.prev = prev;
let (prev, next) = get_links(&response)?;
self.next = next;
self.prev = prev;
Ok(Some(response.json().await?))
Ok(Some(response))
}
Err(err) => {
error!(
err = as_debug!(err), url = url,
method = stringify!($method),
call_id = as_debug!(self.call_id);
"error making API request"
);
Err(err.into())
}
}
});
)*
}
@ -60,33 +90,41 @@ macro_rules! pages {
/// Represents a single page of API results
#[derive(Debug, Clone)]
pub struct Page<T: for<'de> Deserialize<'de>> {
pub struct Page<T: for<'de> Deserialize<'de> + Serialize> {
mastodon: Mastodon,
next: Option<Url>,
prev: Option<Url>,
/// Initial set of items
pub initial_items: Vec<T>,
pub(crate) call_id: Uuid,
}
impl<'a, T: for<'de> Deserialize<'de>> Page<T> {
impl<'a, T: for<'de> Deserialize<'de> + Serialize> Page<T> {
pages! {
next: next_page,
prev: prev_page
}
/// Create a new Page.
pub(crate) async fn new(mastodon: Mastodon, response: Response) -> Result<Self> {
let (prev, next) = get_links(&response)?;
pub(crate) async fn new(mastodon: Mastodon, response: Response, call_id: Uuid) -> Result<Self> {
let (prev, next) = get_links(&response, call_id)?;
let initial_items = response.json().await?;
debug!(
initial_items = as_serde!(initial_items), prev = as_debug!(prev),
next = as_debug!(next), call_id = as_debug!(call_id);
"received first page from API call"
);
Ok(Page {
initial_items: response.json().await?,
initial_items,
next,
prev,
mastodon,
call_id,
})
}
}
impl<T: Clone + for<'de> Deserialize<'de>> Page<T> {
impl<T: Clone + for<'de> Deserialize<'de> + Serialize> Page<T> {
/// Returns an iterator that provides a stream of `T`s
///
/// This abstracts away the process of iterating over each item in a page,
@ -119,12 +157,13 @@ impl<T: Clone + for<'de> Deserialize<'de>> Page<T> {
}
}
fn get_links(response: &Response) -> Result<(Option<Url>, Option<Url>)> {
fn get_links(response: &Response, call_id: Uuid) -> Result<(Option<Url>, Option<Url>)> {
let mut prev = None;
let mut next = None;
if let Some(link_header) = response.headers().get(LINK) {
let link_header = link_header.to_str()?;
trace!(link_header = link_header, call_id = as_debug!(call_id); "parsing link header");
let link_header = link_header.as_bytes();
let link_header: Link = parsing::from_raw_str(&link_header)?;
for value in link_header.values() {
@ -132,6 +171,7 @@ fn get_links(response: &Response) -> Result<(Option<Url>, Option<Url>)> {
if relations.contains(&RelationType::Next) {
// next = Some(Url::parse(value.link())?);
next = if let Ok(url) = Url::parse(value.link()) {
trace!(next = as_debug!(url), call_id = as_debug!(call_id); "parsed link header");
Some(url)
} else {
// HACK: url::ParseError::into isn't working for some reason.
@ -141,6 +181,7 @@ fn get_links(response: &Response) -> Result<(Option<Url>, Option<Url>)> {
if relations.contains(&RelationType::Prev) {
prev = if let Ok(url) = Url::parse(value.link()) {
trace!(prev = as_debug!(url), call_id = as_debug!(call_id); "parsed link header");
Some(url)
} else {
// HACK: url::ParseError::into isn't working for some reason.

51
src/registration.rs
View File

@ -1,10 +1,13 @@
use std::borrow::Cow;
use log::{as_debug, as_serde, debug, error, trace};
use reqwest::Client;
use url::percent_encoding::{utf8_percent_encode, DEFAULT_ENCODE_SET};
use uuid::Uuid;
use crate::{
apps::{App, AppBuilder},
log_serde,
scopes::Scopes,
Data,
Error,
@ -24,7 +27,7 @@ pub struct Registration<'a> {
force_login: bool,
}
#[derive(Deserialize)]
#[derive(Serialize, Deserialize)]
struct OAuth {
client_id: String,
client_secret: String,
@ -36,7 +39,7 @@ fn default_redirect_uri() -> String {
DEFAULT_REDIRECT_URI.to_string()
}
#[derive(Deserialize)]
#[derive(Serialize, Deserialize)]
struct AccessToken {
access_token: String,
}
@ -182,8 +185,30 @@ impl<'a> Registration<'a> {
async fn send_app(&self, app: &App) -> Result<OAuth> {
let url = format!("{}/api/v1/apps", self.base);
let call_id = Uuid::new_v4();
debug!(url = url, app = as_serde!(app), call_id = as_debug!(call_id); "registering app");
let response = self.client.post(&url).json(&app).send().await?;
Ok(response.json().await?)
match response.error_for_status() {
Ok(response) => {
let response = response.json().await?;
debug!(
response = as_serde!(response), app = as_serde!(app),
url = url, method = stringify!($method),
call_id = as_debug!(call_id);
"received API response"
);
Ok(response)
},
Err(err) => {
error!(
err = as_debug!(err), url = url, method = stringify!($method),
call_id = as_debug!(call_id);
"error making API request"
);
Err(err.into())
},
}
}
}
@ -319,17 +344,17 @@ impl Registered {
redirect_uri={}",
self.base, self.client_id, self.client_secret, code, self.redirect
);
let token: AccessToken = self
.client
.post(&url)
.send()
.await?
.error_for_status()?
.json()
.await?;
debug!(url = url; "completing registration");
let response = self.client.post(&url).send().await?;
debug!(
status = log_serde!(response Status), url = url,
headers = log_serde!(response Headers);
"received API response"
);
let token: AccessToken = response.json().await?;
debug!(url = url, body = as_serde!(token); "parsed response body");
let data = self.registered(token.access_token);
trace!(auth_data = as_serde!(data); "registered");
Ok(Mastodon::new(self.client.clone(), data))
}