Migration to 0.3 & Async/Await

crates/ilp-node/Cargo.toml

@@ -24,34 +24,37 @@ required-features = ["redis"]
 
 [dependencies]
 bytes = { version = "0.4.12", default-features = false }
+bytes05 = { package = "bytes", version = "0.5", default-features = false }
 clap = { version = "2.33.0", default-features = false }
 config = { version = "0.9.3", default-features = false, features = ["json", "toml", "yaml"] }
-futures = { version = "0.1.29", default-features = false }
+futures = { version = "0.3.1", default-features = false, features = ["compat"] }
 hex = { version = "0.4.0", default-features = false }
 interledger = { path = "../interledger", version = "^0.6.0", default-features = false, features = ["node"] }
 lazy_static = { version = "1.4.0", default-features = false }
 metrics = { version = "0.12.0", default-features = false, features = ["std"] }
 metrics-core = { version = "0.5.1", default-features = false }
 metrics-runtime = { version = "0.12.0", default-features = false, features = ["metrics-observer-prometheus"] }
 num-bigint = { version = "0.2.3", default-features = false, features = ["std"] }
-redis_crate = { package = "redis", version = "0.13.0", default-features = false, features = ["executor"], optional = true }
+# redis_crate = { package = "redis", version = "0.13.0", default-features = false, features = ["executor"], optional = true }
+redis_crate = { package = "redis", git = "https://github.com/mitsuhiko/redis-rs", optional = true, features = ["tokio-rt-core"] }
 ring = { version = "0.16.9", default-features = false }
 serde = { version = "1.0.101", default-features = false }
-tokio = { version = "0.1.22", default-features = false }
+tokio = { version = "0.2.8", features = ["rt-core", "macros", "time"] }
 tracing = { version = "0.1.9", default-features = true, features = ["log"] }
-tracing-futures = { version = "0.1.1", default-features = true, features = ["tokio", "futures-01"] }
+tracing-futures = { version = "0.2", default-features = true, features = ["tokio", "futures-03"] }
 tracing-subscriber = { version = "0.1.6", default-features = true, features = ["tracing-log"] }
 url = { version = "2.1.0", default-features = false }
 libc = { version = "0.2.62", default-features = false }
-warp = { version = "0.1.20", default-features = false, features = ["websocket"] }
+# warp = { version = "0.1.20", default-features = false, features = ["websocket"] }
+warp = { git = "https://github.com/seanmonstar/warp", default-features = false }
 secrecy = { version = "0.5.1", default-features = false, features = ["alloc", "serde"] }
 uuid = { version = "0.8.1", default-features = false}
 
 # For google-pubsub
 base64 = { version = "0.10.1", default-features = false, optional = true }
 chrono = { version = "0.4.9", default-features = false, features = [], optional = true}
 parking_lot = { version = "0.9.0", default-features = false, optional = true }
-reqwest = { version = "0.9.22", default-features = false, features = ["default-tls"], optional = true }
+reqwest = { version = "0.10.0", default-features = false, features = ["default-tls", "json"], optional = true }
 serde_json = { version = "1.0.41", default-features = false, optional = true }
 yup-oauth2 = { version = "3.1.1", default-features = false, optional = true }
 
@@ -60,7 +63,7 @@ approx = { version = "0.3.2", default-features = false }
 base64 = { version = "0.10.1", default-features = false }
 net2 = { version = "0.2.33", default-features = false }
 rand = { version = "0.7.2", default-features = false }
-reqwest = { version = "0.9.22", default-features = false, features = ["default-tls"] }
+reqwest = { version = "0.10.0", default-features = false, features = ["default-tls", "json"] }
 serde_json = { version = "1.0.41", default-features = false }
 tokio-retry = { version = "0.2.0", default-features = false }
 

crates/ilp-node/src/google_pubsub.rs

@@ -1,15 +1,18 @@
+#[cfg(feature = "google_pubsub")]
 use base64;
 use chrono::Utc;
 use futures::{
+    compat::Future01CompatExt,
     future::{ok, Either},
-    Future,
+    Future, TryFutureExt,
 };
 use interledger::{
+    ccp::CcpRoutingAccount,
     packet::Address,
-    service::{Account, BoxedIlpFuture, OutgoingRequest, OutgoingService, Username},
+    service::{Account, IlpResult, OutgoingRequest, OutgoingService, Username},
 };
 use parking_lot::Mutex;
-use reqwest::r#async::Client;
+use reqwest::Client;
 use serde::{Deserialize, Serialize};
 use std::{collections::HashMap, sync::Arc};
 use tokio::spawn;
@@ -58,16 +61,17 @@ struct PacketRecord {
     timestamp: String,
 }
 
+use std::pin::Pin;
 /// Create an Interledger service wrapper that publishes records
 /// of fulfilled packets to Google Cloud PubSub.
 ///
 /// This is an experimental feature that may be removed in the future.
-pub fn create_google_pubsub_wrapper<
+pub async fn create_google_pubsub_wrapper<
     A: Account + 'static,
     O: OutgoingService<A> + Clone + Send + 'static,
 >(
     config: Option<PubsubConfig>,
-) -> impl Fn(OutgoingRequest<A>, O) -> BoxedIlpFuture + Clone {
+) -> impl Fn(OutgoingRequest<A>, O) -> Pin<Box<Future<Output = IlpResult>>> + Clone {
     // If Google credentials were passed in, create an HTTP client and
     // OAuth2 client that will automatically fetch and cache access tokens
     let utilities = if let Some(config) = config {
@@ -91,10 +95,16 @@ pub fn create_google_pubsub_wrapper<
         None
     };
 
-    move |request: OutgoingRequest<A>, mut next: O| -> BoxedIlpFuture {
+    move |request: OutgoingRequest<A>, mut next: O| -> Pin<Box<Future<Output = IlpResult>>> {
+        let mut next_clone = next.clone();
+        let mut next_clone2 = next.clone();
         match &utilities {
             // Just pass the request on if no Google Pubsub details were configured
-            None => Box::new(next.send_request(request)),
+            // Due to using async_trait this becomes a Box::pin!
+            None => Box::pin(async move {
+                let fulfill = next_clone.send_request(request).await?;
+                Ok(fulfill)
+            }),
             Some((client, api_endpoint, token_fetcher)) => {
                 let prev_hop_account = request.from.username().clone();
                 let prev_hop_asset_code = request.from.asset_code().to_string();
@@ -108,78 +118,80 @@ pub fn create_google_pubsub_wrapper<
                 let client = client.clone();
                 let api_endpoint = api_endpoint.clone();
                 let token_fetcher = token_fetcher.clone();
-
-                Box::new(next.send_request(request).map(move |fulfill| {
+                Box::pin(async move {
                     // Only fulfilled packets are published for now
+                    let fulfill = next_clone2.send_request(request).await?;
                     let fulfillment = base64::encode(fulfill.fulfillment());
 
-                    let get_token_future = token_fetcher.lock()
+                    let get_token_future = token_fetcher
+                        .lock()
                         .token(TOKEN_SCOPES)
+                        .compat()
+                        .map_ok(|token: yup_oauth2::Token| token.access_token)
                         .map_err(|err| {
                             error!("Error fetching OAuth token for Google PubSub: {:?}", err)
                         });
+
                     // Spawn a task to submit the packet to PubSub so we
                     // don't block returning the fulfillment
                     // Note this means that if there is a problem submitting the
                     // packet record to PubSub, it will only log an error
-                    spawn(
-                        get_token_future
-                            .and_then(move |token| {
-                                let record = PacketRecord {
-                                    prev_hop_account,
-                                    prev_hop_asset_code,
-                                    prev_hop_asset_scale,
-                                    prev_hop_amount,
-                                    next_hop_account,
-                                    next_hop_asset_code,
-                                    next_hop_asset_scale,
-                                    next_hop_amount,
-                                    destination_ilp_address,
-                                    fulfillment,
-                                    timestamp: Utc::now().to_rfc3339(),
-                                };
-                                let data = base64::encode(&serde_json::to_string(&record).unwrap());
+                    spawn(async move {
+                        let token = get_token_future.await?;
+
+                        let record = PacketRecord {
+                            prev_hop_account,
+                            prev_hop_asset_code,
+                            prev_hop_asset_scale,
+                            prev_hop_amount,
+                            next_hop_account,
+                            next_hop_asset_code,
+                            next_hop_asset_scale,
+                            next_hop_amount,
+                            destination_ilp_address,
+                            fulfillment,
+                            timestamp: Utc::now().to_rfc3339(),
+                        };
+                        let data = base64::encode(&serde_json::to_string(&record).unwrap());
+
+                        let res = client
+                            .post(api_endpoint.as_str())
+                            .bearer_auth(token)
+                            .json(&PubsubRequest {
+                                messages: vec![PubsubMessage {
+                                    // TODO should there be an ID?
+                                    message_id: None,
+                                    data: Some(data),
+                                    attributes: None,
+                                    publish_time: None,
+                                }],
+                            })
+                            .send()
+                            .map_err(|err| {
+                                error!("Error sending packet details to Google PubSub: {:?}", err)
+                            })
+                            .await?;
+
+                        if res.status().is_success() {
+                            return Ok(());
+                        } else {
+                            let status = res.status();
+                            let body = res
+                                .text()
+                                .map_err(|err| error!("Error getting response body: {:?}", err))
+                                .await?;
+                            error!(
+                                %status,
+                                "Error sending packet details to Google PubSub: {}",
+                                body
+                            );
+                        }
+
+                        Ok::<(), ()>(())
+                    });
 
-                                client
-                                    .post(api_endpoint.as_str())
-                                    .bearer_auth(token.access_token)
-                                    .json(&PubsubRequest {
-                                        messages: vec![PubsubMessage {
-                                            // TODO should there be an ID?
-                                            message_id: None,
-                                            data: Some(data),
-                                            attributes: None,
-                                            publish_time: None,
-                                        }],
-                                    })
-                                    .send()
-                                    .map_err(|err| {
-                                        error!(
-                                            "Error sending packet details to Google PubSub: {:?}",
-                                            err
-                                        )
-                                    })
-                                    .and_then(|mut res| {
-                                        if res.status().is_success() {
-                                            Either::A(ok(()))
-                                        } else {
-                                            let status = res.status();
-                                            Either::B(res.text()
-                                                .map_err(|err| error!("Error getting response body: {:?}", err))
-                                                .and_then(move |body| {
-                                                    error!(
-                                                        %status,
-                                                        "Error sending packet details to Google PubSub: {}",
-                                                        body
-                                                    );
-                                                    Ok(())
-                                                }))
-                                        }
-                                    })
-                            }),
-                    );
-                    fulfill
-                }))
+                    Ok(fulfill)
+                })
             }
         }
     }

crates/ilp-node/src/lib.rs

@@ -1,15 +1,12 @@
-#![type_length_limit = "1152909"]
+#![type_length_limit = "6000000"]
 
-mod metrics;
+// mod metrics;
 mod node;
-mod trace;
+// mod trace;
 
-#[cfg(feature = "google-pubsub")]
-mod google_pubsub;
+// #[cfg(feature = "google-pubsub")]
+// mod google_pubsub;
 #[cfg(feature = "redis")]
 mod redis_store;
 
 pub use node::*;
-#[allow(deprecated)]
-#[cfg(feature = "redis")]
-pub use redis_store::insert_account_with_redis_store;

crates/ilp-node/src/main.rs

@@ -1,4 +1,4 @@
-#![type_length_limit = "1152909"]
+#![type_length_limit = "6000000"]
 
 mod metrics;
 mod node;
@@ -24,7 +24,8 @@ use tracing_subscriber::{
     fmt::{time::ChronoUtc, Subscriber},
 };
 
-pub fn main() {
+#[tokio::main]
+async fn main() {
     Subscriber::builder()
         .with_timer(ChronoUtc::rfc3339())
         .with_env_filter(EnvFilter::from_default_env())
@@ -143,7 +144,9 @@ pub fn main() {
     }
     let matches = app.clone().get_matches();
     merge_args(&mut config, &matches);
-    config.try_into::<InterledgerNode>().unwrap().run();
+
+    let node = config.try_into::<InterledgerNode>().unwrap();
+    node.serve().await;
 }
 
 // returns (subcommand paths, config path)