eth-engine: implement LeftoversStore for tests

Author: gakonst

crates/interledger-settlement-engines/src/engines/ethereum_ledger/test_helpers.rs

@@ -61,6 +61,19 @@ pub struct TestStore {
     pub last_observed_block: Arc<RwLock<U256>>,
     pub saved_hashes: Arc<RwLock<HashMap<H256, bool>>>,
     pub cache_hits: Arc<RwLock<u64>>,
+    pub leftovers: Arc<RwLock<HashMap<String, String>>>,
+}
+
+use crate::stores::LeftoversStore;
+use num_bigint::BigUint;
+impl LeftoversStore for TestStore {
+    fn save_leftovers(
+        &self,
+        account_id: String,
+        leftovers: BigUint,
+    ) -> Box<Future<Item = (), Error = ()> + Send> {
+        Box::new(ok(()))
+    }
 }
 
 impl EthereumStore for TestStore {
@@ -236,6 +249,7 @@ impl TestStore {
             cache_hits: Arc::new(RwLock::new(0)),
             last_observed_block: Arc::new(RwLock::new(U256::from(0))),
             saved_hashes: Arc::new(RwLock::new(HashMap::new())),
+            leftovers: Arc::new(RwLock::new(HashMap::new())),
         }
     }
 }
@@ -263,7 +277,13 @@ pub fn test_engine<Si, S, A>(
 ) -> EthereumLedgerSettlementEngine<S, Si, A>
 where
     Si: EthereumLedgerTxSigner + Clone + Send + Sync + 'static,
-    S: EthereumStore<Account = A> + IdempotentStore + Clone + Send + Sync + 'static,
+    S: EthereumStore<Account = A>
+        + LeftoversStore
+        + IdempotentStore
+        + Clone
+        + Send
+        + Sync
+        + 'static,
     A: EthereumAccount + Send + Sync + 'static,
 {
     EthereumLedgerSettlementEngineBuilder::new(store, key)

crates/interledger-settlement-engines/src/stores/redis_ethereum_ledger/store.rs

@@ -109,7 +109,16 @@ impl EthereumLedgerRedisStore {
 
 impl LeftoversStore for EthereumLedgerRedisStore {
     fn save_leftovers(&self, account_id: String, leftovers: BigUint) -> Box<dyn Future<Item = (), Error = ()> + Send> {
-        Box::new(ok(()))
+        let mut pipe = redis::pipe();
+        pipe.set(format!("leftovers:{}", account_id), leftovers.to_string())
+            .ignore();
+        Box::new(
+            pipe.query_async(self.connection.clone())
+                .map_err(move |err| {
+                    error!("Error saving leftovers {:?}: {:?}", leftovers, err)
+                })
+                .and_then(move |(_conn, _ret): (_, Value)| Ok(())),
+        )
     }
 }