Merge #1733

1733: Refactor Spawn r=tcharding a=tcharding

The `Spawn` trait is implemented on `LedgerConnectors`, there are a
bunch of refactorings that can be done, this patch does them all.

Some highlights are:
- Move `http_api::Dependencies` to `crate::Facade`
- Inline `LedgerConnectors` fields into `Facade`
- Create `init_accepted_swap()` that does the state store insert, initiates the state_machine for this swap, and spawns the swap

Resolves: #1679

Co-authored-by: Tobin C. Harding <tobin.harding@coblox.tech>

Author: bors[bot]

cnd/src/http_api/mod.rs

@@ -3,13 +3,11 @@ pub mod routes;
 #[macro_use]
 pub mod impl_serialize_http;
 pub mod action;
-mod dependencies;
 mod ethereum_network;
 mod problem;
 mod swap_resource;
 
 pub use self::{
-    dependencies::*,
     problem::*,
     swap_resource::{SwapParameters, SwapResource, SwapStatus},
 };

cnd/src/http_api/route_factory.rs

@@ -4,13 +4,10 @@ use crate::{
     http_api,
     network::{Network, SendRequest},
     seed::SwapSeed,
-    swap_protocols::{
-        self,
-        rfc003::{state_store::StateStore, Spawn},
-        SwapId,
-    },
+    swap_protocols::{self, rfc003::state_store::StateStore, LedgerEventsCreator, SwapId},
 };
 use libp2p::PeerId;
+use tokio::executor::Executor;
 use warp::{self, filters::BoxedFilter, Filter, Reply};
 
 pub const RFC003: &str = "rfc003";
@@ -26,12 +23,13 @@ pub fn new_action_link(id: &SwapId, action: &str) -> String {
 pub fn create<
     D: Clone
         + StateStore
+        + Executor
         + Network
         + SendRequest
-        + Spawn
         + SwapSeed
         + DetermineTypes
         + Retrieve
+        + LedgerEventsCreator
         + Saver,
 >(
     peer_id: PeerId,

cnd/src/http_api/routes/rfc003/handlers/action.rs

@@ -12,26 +12,35 @@ use crate::{
     network::Network,
     seed::SwapSeed,
     swap_protocols::{
+        self,
         actions::Actions,
         rfc003::{
             self,
             actions::{Action, ActionKind},
             bob::State,
             messages::{Decision, IntoAcceptMessage},
             state_store::StateStore,
-            Spawn,
         },
-        SwapId,
+        LedgerEventsCreator, SwapId,
     },
 };
 use anyhow::Context;
-use futures::Stream;
 use libp2p_comit::frame::Response;
 use std::fmt::Debug;
+use tokio::executor::Executor;
 use warp::http;
 
 #[allow(clippy::unit_arg, clippy::let_unit_value, clippy::cognitive_complexity)]
-pub async fn handle_action<D: StateStore + Network + Spawn + SwapSeed + Saver + DetermineTypes>(
+pub async fn handle_action<
+    D: StateStore
+        + Network
+        + SwapSeed
+        + Saver
+        + DetermineTypes
+        + LedgerEventsCreator
+        + Executor
+        + Clone,
+>(
     method: http::Method,
     swap_id: SwapId,
     action_kind: ActionKind,
@@ -76,16 +85,12 @@ pub async fn handle_action<D: StateStore + Network + Spawn + SwapSeed + Saver +
                 })?;
 
                 let swap_request = state.request();
-                let seed = dependencies.swap_seed(swap_id);
-                let state = State::accepted(swap_request.clone(), accept_message, seed);
-                StateStore::insert(&dependencies, swap_id, state);
-
-                let receiver = Spawn::spawn(&dependencies, swap_request, accept_message);
-
-                tokio::spawn(receiver.for_each(move |update| {
-                    StateStore::update::<State<AL, BL, AA, BA>>(&dependencies, &swap_id, update);
-                    Ok(())
-                }));
+                swap_protocols::init_accepted_swap(
+                    &dependencies,
+                    swap_request,
+                    accept_message,
+                    types.role,
+                )?;
 
                 Ok(ActionResponseBody::None)
             }

cnd/src/http_api/routes/rfc003/handlers/post_swap.rs

@@ -5,26 +5,38 @@ use crate::{
     network::{DialInformation, SendRequest},
     seed::SwapSeed,
     swap_protocols::{
+        self,
         asset::Asset,
         ledger,
         rfc003::{
-            self, alice::State, create_ledger_events::CreateLedgerEvents, state_store::StateStore,
-            Accept, Ledger, Request, SecretHash, SecretSource, Spawn,
+            self, alice::State, state_store::StateStore, Accept, Decline, Ledger, Request,
+            SecretHash, SecretSource,
         },
-        HashFunction, LedgerConnectors, Role, SwapId,
+        HashFunction, LedgerEventsCreator, Role, SwapId,
     },
     timestamp::Timestamp,
+    CreateLedgerEvents,
 };
-use futures::Stream;
+use anyhow::Context;
+use futures::Future;
 use futures_core::{
     compat::Future01CompatExt,
     future::{FutureExt, TryFutureExt},
 };
 use serde::{Deserialize, Serialize};
 use std::str::FromStr;
+use tokio::executor::Executor;
 
 pub async fn handle_post_swap<
-    D: Clone + StateStore + Save<Swap> + SendRequest + Spawn + SwapSeed + Saver,
+    D: Clone
+        + Executor
+        + StateStore
+        + Save<Swap>
+        + SendRequest
+        + SwapSeed
+        + Saver
+        + Clone
+        + LedgerEventsCreator,
 >(
     dependencies: D,
     body: serde_json::Value,
@@ -196,14 +208,18 @@ async fn initiate_request<D, AL, BL, AA, BA>(
     swap_request: rfc003::Request<AL, BL, AA, BA>,
 ) -> anyhow::Result<()>
 where
-    LedgerConnectors: CreateLedgerEvents<AL, AA> + CreateLedgerEvents<BL, BA>,
     D: StateStore
+        + Executor
         + SendRequest
-        + Spawn
         + SwapSeed
-        + Saver
         + Save<Request<AL, BL, AA, BA>>
-        + Save<Accept<AL, BL>>,
+        + Save<Accept<AL, BL>>
+        + Save<Swap>
+        + Save<Decline>
+        + LedgerEventsCreator
+        + CreateLedgerEvents<AL, AA>
+        + CreateLedgerEvents<BL, BA>
+        + Clone,
     AL: Ledger,
     BL: Ledger,
     AA: Asset,
@@ -224,41 +240,32 @@ where
                 .send_request(peer.clone(), swap_request.clone())
                 .compat()
                 .await
-                .map_err(|e| {
-                    log::error!(
-                        "Failed to send swap request to {} because {:?}",
-                        peer.clone(),
-                        e
-                    );
-                })?;
+                .with_context(|| format!("Failed to send swap request to {}", peer.clone()))?;
 
             match response {
                 Ok(accept) => {
-                    let state = State::accepted(swap_request.clone(), accept, seed);
-                    StateStore::insert(&dependencies, id, state.clone());
-                    Save::save(&dependencies, accept)
-                        .await
-                        .expect("failed to save message to db");
-
-                    let receiver = Spawn::spawn(&dependencies, swap_request, accept);
-                    tokio::spawn(receiver.for_each(move |update| {
-                        StateStore::update::<State<AL, BL, AA, BA>>(&dependencies, &id, update);
-                        Ok(())
-                    }));
+                    Save::save(&dependencies, accept).await?;
+
+                    swap_protocols::init_accepted_swap(
+                        &dependencies,
+                        swap_request,
+                        accept,
+                        Role::Alice,
+                    )?;
                 }
                 Err(decline) => {
                     log::info!("Swap declined: {:?}", decline);
                     let state = State::declined(swap_request.clone(), decline.clone(), seed);
                     StateStore::insert(&dependencies, id, state.clone());
-                    Save::save(&dependencies, decline.clone())
-                        .await
-                        .expect("failed to save message to db");
+                    Save::save(&dependencies, decline.clone()).await?;
                 }
             };
             Ok(())
         }
     };
-    tokio::spawn(future.boxed().compat());
+    tokio::spawn(future.boxed().compat().map_err(|e: anyhow::Error| {
+        log::error!("{:?}", e);
+    }));
     Ok(())
 }
 

cnd/src/http_api/routes/rfc003/mod.rs

@@ -16,8 +16,8 @@ use crate::{
     network::{Network, SendRequest},
     seed::SwapSeed,
     swap_protocols::{
-        rfc003::{actions::ActionKind, state_store::StateStore, Spawn},
-        SwapId,
+        rfc003::{actions::ActionKind, state_store::StateStore},
+        LedgerEventsCreator, SwapId,
     },
 };
 use futures::Future;
@@ -27,9 +27,19 @@ use warp::{http, Rejection, Reply};
 
 pub use self::swap_state::{LedgerState, SwapCommunication, SwapCommunicationState, SwapState};
 use crate::{db::Saver, http_api::problem};
+use tokio::executor::Executor;
 
 #[allow(clippy::needless_pass_by_value)]
-pub fn post_swap<D: Clone + StateStore + Save<Swap> + SendRequest + Spawn + SwapSeed + Saver>(
+pub fn post_swap<
+    D: Clone
+        + StateStore
+        + Executor
+        + Save<Swap>
+        + SendRequest
+        + SwapSeed
+        + Saver
+        + LedgerEventsCreator,
+>(
     dependencies: D,
     body: serde_json::Value,
 ) -> impl Future<Item = impl Reply, Error = Rejection> {
@@ -60,7 +70,17 @@ pub fn get_swap<D: DetermineTypes + Retrieve + StateStore>(
 }
 
 #[allow(clippy::needless_pass_by_value)]
-pub fn action<D: DetermineTypes + Retrieve + StateStore + Network + Spawn + SwapSeed + Saver>(
+pub fn action<
+    D: DetermineTypes
+        + Retrieve
+        + StateStore
+        + Executor
+        + Clone
+        + Network
+        + SwapSeed
+        + Saver
+        + LedgerEventsCreator,
+>(
     method: http::Method,
     id: SwapId,
     action_kind: ActionKind,

cnd/src/lib.rs

@@ -35,6 +35,10 @@ pub mod spectral_ext;
 pub mod swap_protocols;
 pub mod timestamp;
 
+use crate::swap_protocols::{
+    asset::Asset,
+    rfc003::{events::LedgerEvents, Ledger},
+};
 use anyhow::Context;
 use directories::ProjectDirs;
 use std::path::{Path, PathBuf};
@@ -63,3 +67,7 @@ pub fn default_config_path() -> anyhow::Result<PathBuf> {
 pub fn data_dir() -> Option<PathBuf> {
     ProjectDirs::from("", "", "comit").map(|proj_dirs| proj_dirs.data_dir().to_path_buf())
 }
+
+pub trait CreateLedgerEvents<L: Ledger, A: Asset> {
+    fn create_ledger_events(&self) -> Box<dyn LedgerEvents<L, A>>;
+}

cnd/src/load_swaps.rs

@@ -1,71 +1,48 @@
 #![allow(clippy::type_repetition_in_bounds)]
 use crate::{
-    db::{AcceptedSwap, DetermineTypes, LoadAcceptedSwap, Retrieve, Sqlite},
-    seed::Seed,
+    db::{DetermineTypes, LoadAcceptedSwap, Retrieve},
+    ethereum::{Erc20Token, EtherQuantity},
+    seed::SwapSeed,
     swap_protocols::{
-        ledger::LedgerConnectors,
-        rfc003::{
-            state_store::{InMemoryStateStore, StateStore},
-            Spawn,
-        },
+        self,
+        ledger::{Bitcoin, Ethereum},
+        rfc003::state_store::StateStore,
+        LedgerEventsCreator,
     },
 };
-use futures::Stream;
-use std::sync::Arc;
+use tokio::executor::Executor;
 
 #[allow(clippy::cognitive_complexity)]
-pub async fn load_swaps_from_database(
-    ledger_events: LedgerConnectors,
-    state_store: Arc<InMemoryStateStore>,
-    seed: Seed,
-    db: Sqlite,
-) -> anyhow::Result<()> {
+pub async fn load_swaps_from_database<D>(dependencies: D) -> anyhow::Result<()>
+where
+    D: StateStore
+        + Executor
+        + Clone
+        + SwapSeed
+        + LedgerEventsCreator
+        + Retrieve
+        + DetermineTypes
+        + LoadAcceptedSwap<Bitcoin, Ethereum, bitcoin::Amount, EtherQuantity>
+        + LoadAcceptedSwap<Ethereum, Bitcoin, EtherQuantity, bitcoin::Amount>
+        + LoadAcceptedSwap<Bitcoin, Ethereum, bitcoin::Amount, Erc20Token>
+        + LoadAcceptedSwap<Ethereum, Bitcoin, Erc20Token, bitcoin::Amount>,
+{
     log::debug!("loading swaps from database ...");
 
-    for swap in db.all().await?.iter() {
+    for swap in Retrieve::all(&dependencies).await?.iter() {
         let swap_id = swap.swap_id;
         log::debug!("got swap from database: {}", swap_id);
 
-        let types = db.determine_types(&swap_id).await?;
+        let types = DetermineTypes::determine_types(&dependencies, &swap_id).await?;
 
         with_swap_types!(types, {
-            let accepted: Result<AcceptedSwap<AL, BL, AA, BA>, anyhow::Error> =
-                db.load_accepted_swap(&swap_id).await;
+            let accepted =
+                LoadAcceptedSwap::<AL, BL, AA, BA>::load_accepted_swap(&dependencies, &swap_id)
+                    .await;
 
             match accepted {
                 Ok((request, accept, _at)) => {
-                    match types.role {
-                        Role::Alice => {
-                            let state = alice::State::accepted(request.clone(), accept, seed);
-                            state_store.insert(swap_id, state);
-
-                            let receiver = ledger_events.spawn(request, accept);
-
-                            tokio::spawn(receiver.for_each({
-                                let state_store = state_store.clone();
-                                move |update| {
-                                    state_store
-                                        .update::<alice::State<AL, BL, AA, BA>>(&swap_id, update);
-                                    Ok(())
-                                }
-                            }));
-                        }
-                        Role::Bob => {
-                            let state = bob::State::accepted(request.clone(), accept, seed);
-                            state_store.insert(swap_id, state);
-
-                            let receiver = ledger_events.spawn(request, accept);
-
-                            tokio::spawn(receiver.for_each({
-                                let state_store = state_store.clone();
-                                move |update| {
-                                    state_store
-                                        .update::<bob::State<AL, BL, AA, BA>>(&swap_id, update);
-                                    Ok(())
-                                }
-                            }));
-                        }
-                    };
+                    swap_protocols::init_accepted_swap(&dependencies, request, accept, types.role)?;
                 }
                 Err(e) => log::error!("failed to load swap: {}, continuing ...", e),
             };