feat(settlement/api): Upgrade the message service

This service is responsible for catching any messages sent to the Settlement ILP Address and forward them to the engine

Author: gakonst

crates/interledger-settlement/src/api/message_service.rs

@@ -1,12 +1,10 @@
 use crate::core::types::{SettlementAccount, SE_ILP_ADDRESS};
-use futures::{
-    future::{err, Either},
-    Future, Stream,
-};
+use async_trait::async_trait;
+use futures::{compat::Future01CompatExt, TryFutureExt};
 use interledger_packet::{ErrorCode, FulfillBuilder, RejectBuilder};
-use interledger_service::{Account, BoxedIlpFuture, IncomingRequest, IncomingService};
+use interledger_service::{Account, IlpResult, IncomingRequest, IncomingService};
 use log::error;
-use reqwest::r#async::Client;
+use reqwest::Client;
 use std::marker::PhantomData;
 use tokio_retry::{strategy::ExponentialBackoff, Retry};
 
@@ -33,14 +31,13 @@ where
     }
 }
 
+#[async_trait]
 impl<I, A> IncomingService<A> for SettlementMessageService<I, A>
 where
     I: IncomingService<A> + Send,
-    A: SettlementAccount + Account,
+    A: SettlementAccount + Account + Send + Sync,
 {
-    type Future = BoxedIlpFuture;
-
-    fn handle_request(&mut self, request: IncomingRequest<A>) -> Self::Future {
+    async fn handle_request(&mut self, request: IncomingRequest<A>) -> IlpResult {
         // Only handle the request if the destination address matches the ILP address
         // of the settlement engine being used for this account
         if let Some(settlement_engine_details) = request.from.settlement_engine_details() {
@@ -67,73 +64,94 @@ where
                         .header("Idempotency-Key", idempotency_uuid.clone())
                         .body(message.clone())
                         .send()
+                        .compat() // Wrap to a 0.1 future
                 };
+                // TODO: futures-retry is still not on futures 0.3. As a result, we wrap our action in a
+                // 0.1 future, and then wrap the Retry future in a 0.3 future to use async/await.
 
-                return Box::new(Retry::spawn(ExponentialBackoff::from_millis(10).take(10), action)
-                .map_err(move |error| {
-                    error!("Error sending message to settlement engine: {:?}", error);
-                    RejectBuilder {
-                        code: ErrorCode::T00_INTERNAL_ERROR,
-                        message: b"Error sending message to settlement engine",
-                        data: &[],
-                        triggered_by: Some(&SE_ILP_ADDRESS),
-                    }.build()
-                })
-                .and_then(move |response| {
-                    let status = response.status();
-                    if status.is_success() {
-                        Either::A(response.into_body().concat2().map_err(move |err| {
-                            error!("Error concatenating settlement engine response body: {:?}", err);
+                let response = Retry::spawn(ExponentialBackoff::from_millis(10).take(10), action)
+                    .compat()
+                    .map_err(move |error| {
+                        error!("Error sending message to settlement engine: {:?}", error);
+                        RejectBuilder {
+                            code: ErrorCode::T00_INTERNAL_ERROR,
+                            message: b"Error sending message to settlement engine",
+                            data: &[],
+                            triggered_by: Some(&SE_ILP_ADDRESS),
+                        }
+                        .build()
+                    })
+                    .await?;
+                let status = response.status();
+                if status.is_success() {
+                    let body = response
+                        .bytes()
+                        .map_err(|err| {
+                            error!(
+                                "Error concatenating settlement engine response body: {:?}",
+                                err
+                            );
                             RejectBuilder {
                                 code: ErrorCode::T00_INTERNAL_ERROR,
                                 message: b"Error getting settlement engine response",
                                 data: &[],
                                 triggered_by: Some(&SE_ILP_ADDRESS),
-                            }.build()
+                            }
+                            .build()
                         })
-                        .and_then(|body| {
-                            Ok(FulfillBuilder {
-                                fulfillment: &PEER_FULFILLMENT,
-                                data: body.as_ref(),
-                            }.build())
-                        }))
+                        .await?;
+
+                    return Ok(FulfillBuilder {
+                        fulfillment: &PEER_FULFILLMENT,
+                        data: body.as_ref(),
+                    }
+                    .build());
+                } else {
+                    error!(
+                        "Settlement engine rejected message with HTTP error code: {}",
+                        response.status()
+                    );
+                    let code = if status.is_client_error() {
+                        ErrorCode::F00_BAD_REQUEST
                     } else {
-                        error!("Settlement engine rejected message with HTTP error code: {}", response.status());
-                        let code = if status.is_client_error() {
-                            ErrorCode::F00_BAD_REQUEST
-                        } else {
-                            ErrorCode::T00_INTERNAL_ERROR
-                        };
-                        Either::B(err(RejectBuilder {
-                            code,
-                            message: format!("Settlement engine rejected request with error code: {}", response.status()).as_str().as_ref(),
-                            data: &[],
-                            triggered_by: Some(&SE_ILP_ADDRESS),
-                        }.build()))
+                        ErrorCode::T00_INTERNAL_ERROR
+                    };
+
+                    return Err(RejectBuilder {
+                        code,
+                        message: format!(
+                            "Settlement engine rejected request with error code: {}",
+                            response.status()
+                        )
+                        .as_str()
+                        .as_ref(),
+                        data: &[],
+                        triggered_by: Some(&SE_ILP_ADDRESS),
                     }
-                }));
+                    .build());
+                }
             }
         }
-        Box::new(self.next.handle_request(request))
+        self.next.handle_request(request).await
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::api::fixtures::{BODY, DATA, SERVICE_ADDRESS, TEST_ACCOUNT_0};
-    use crate::api::test_helpers::{block_on, mock_message, test_service};
+    use crate::api::test_helpers::{mock_message, test_service};
     use interledger_packet::{Address, Fulfill, PrepareBuilder, Reject};
     use std::str::FromStr;
     use std::time::SystemTime;
 
-    #[test]
-    fn settlement_ok() {
+    #[tokio::test]
+    async fn settlement_ok() {
         // happy case
         let m = mock_message(200).create();
         let mut settlement = test_service();
-        let fulfill: Fulfill = block_on(
-            settlement.handle_request(IncomingRequest {
+        let fulfill: Fulfill = settlement
+            .handle_request(IncomingRequest {
                 from: TEST_ACCOUNT_0.clone(),
                 prepare: PrepareBuilder {
                     amount: 0,
@@ -143,22 +161,22 @@ mod tests {
                     execution_condition: &[0; 32],
                 }
                 .build(),
-            }),
-        )
-        .unwrap();
+            })
+            .await
+            .unwrap();
 
         m.assert();
         assert_eq!(fulfill.data(), BODY.as_bytes());
         assert_eq!(fulfill.fulfillment(), &[0; 32]);
     }
 
-    #[test]
-    fn gets_forwarded_if_destination_not_engine_() {
+    #[tokio::test]
+    async fn gets_forwarded_if_destination_not_engine_() {
         let m = mock_message(200).create().expect(0);
         let mut settlement = test_service();
         let destination = Address::from_str("example.some.address").unwrap();
-        let reject: Reject = block_on(
-            settlement.handle_request(IncomingRequest {
+        let reject: Reject = settlement
+            .handle_request(IncomingRequest {
                 from: TEST_ACCOUNT_0.clone(),
                 prepare: PrepareBuilder {
                     amount: 0,
@@ -168,24 +186,24 @@ mod tests {
                     execution_condition: &[0; 32],
                 }
                 .build(),
-            }),
-        )
-        .unwrap_err();
+            })
+            .await
+            .unwrap_err();
 
         m.assert();
         assert_eq!(reject.code(), ErrorCode::F02_UNREACHABLE);
         assert_eq!(reject.triggered_by().unwrap(), SERVICE_ADDRESS.clone());
         assert_eq!(reject.message(), b"No other incoming handler!" as &[u8],);
     }
 
-    #[test]
-    fn account_does_not_have_settlement_engine() {
+    #[tokio::test]
+    async fn account_does_not_have_settlement_engine() {
         let m = mock_message(200).create().expect(0);
         let mut settlement = test_service();
         let mut acc = TEST_ACCOUNT_0.clone();
         acc.no_details = true; // Hide the settlement engine data from the account
-        let reject: Reject = block_on(
-            settlement.handle_request(IncomingRequest {
+        let reject: Reject = settlement
+            .handle_request(IncomingRequest {
                 from: acc.clone(),
                 prepare: PrepareBuilder {
                     amount: 0,
@@ -195,25 +213,25 @@ mod tests {
                     execution_condition: &[0; 32],
                 }
                 .build(),
-            }),
-        )
-        .unwrap_err();
+            })
+            .await
+            .unwrap_err();
 
         m.assert();
         assert_eq!(reject.code(), ErrorCode::F02_UNREACHABLE);
         assert_eq!(reject.triggered_by().unwrap(), SERVICE_ADDRESS.clone());
         assert_eq!(reject.message(), b"No other incoming handler!");
     }
 
-    #[test]
-    fn settlement_engine_rejects() {
+    #[tokio::test]
+    async fn settlement_engine_rejects() {
         // for whatever reason the engine rejects our request with a 500 code
         let error_code = 500;
         let error_str = "Internal Server Error";
         let m = mock_message(error_code).create();
         let mut settlement = test_service();
-        let reject: Reject = block_on(
-            settlement.handle_request(IncomingRequest {
+        let reject: Reject = settlement
+            .handle_request(IncomingRequest {
                 from: TEST_ACCOUNT_0.clone(),
                 prepare: PrepareBuilder {
                     amount: 0,
@@ -223,9 +241,9 @@ mod tests {
                     execution_condition: &[0; 32],
                 }
                 .build(),
-            }),
-        )
-        .unwrap_err();
+            })
+            .await
+            .unwrap_err();
 
         m.assert();
         assert_eq!(reject.code(), ErrorCode::T00_INTERNAL_ERROR);