Merge pull request #43 from Overmuse/SR/refactoring

refactor

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "order-manager"
-version = "7.0.2"
+version = "7.0.3"
 authors = ["Sebastian Rollen <seb@overmu.se>"]
 edition = "2018"
 

src/db/allocations.rs

@@ -1,12 +1,11 @@
 use crate::types::{Allocation, Owner};
 use std::convert::TryInto;
-use std::sync::Arc;
-use tokio_postgres::{Client, Error};
+use tokio_postgres::{Error, GenericClient};
 use tracing::trace;
 use uuid::Uuid;
 
 #[tracing::instrument(skip(client))]
-pub async fn get_allocations(client: Arc<Client>) -> Result<Vec<Allocation>, Error> {
+pub async fn get_allocations<T: GenericClient>(client: &T) -> Result<Vec<Allocation>, Error> {
     trace!("Getting allocations");
     client
         .query("SELECT * FROM allocations", &[])
@@ -16,8 +15,8 @@ pub async fn get_allocations(client: Arc<Client>) -> Result<Vec<Allocation>, Err
         .collect()
 }
 
-pub async fn set_allocation_owner(
-    client: Arc<Client>,
+pub async fn set_allocation_owner<T: GenericClient>(
+    client: &T,
     id: Uuid,
     owner: Owner,
 ) -> Result<(), Error> {
@@ -48,7 +47,10 @@ pub async fn set_allocation_owner(
 }
 
 #[tracing::instrument(skip(client, allocation))]
-pub async fn save_allocation(client: Arc<Client>, allocation: Allocation) -> Result<(), Error> {
+pub async fn save_allocation<T: GenericClient>(
+    client: &T,
+    allocation: Allocation,
+) -> Result<(), Error> {
     trace!("Saving allocation");
     let (owner, sub_owner) = match allocation.owner {
         Owner::House => ("House".to_string(), None),

src/db/claims.rs

@@ -1,14 +1,12 @@
 use super::utils::split_amount_spec;
 use crate::types::Claim;
 use std::convert::TryInto;
-use std::sync::Arc;
-use tokio_postgres::{Client, Error};
+use tokio_postgres::{Error, GenericClient};
 use tracing::trace;
 use trading_base::AmountSpec;
 use uuid::Uuid;
 
-#[tracing::instrument(skip(client))]
-pub async fn get_claims(client: Arc<Client>) -> Result<Vec<Claim>, Error> {
+pub async fn get_claims<T: GenericClient>(client: &T) -> Result<Vec<Claim>, Error> {
     trace!("Getting claims");
     client
         .query("SELECT * FROM claims", &[])
@@ -19,7 +17,10 @@ pub async fn get_claims(client: Arc<Client>) -> Result<Vec<Claim>, Error> {
 }
 
 #[tracing::instrument(skip(client, ticker))]
-pub async fn get_claims_by_ticker(client: Arc<Client>, ticker: &str) -> Result<Vec<Claim>, Error> {
+pub async fn get_claims_by_ticker<T: GenericClient>(
+    client: &T,
+    ticker: &str,
+) -> Result<Vec<Claim>, Error> {
     trace!(ticker, "Getting claims");
     client
         .query("SELECT * FROM claims WHERE ticker = $1", &[&ticker])
@@ -30,7 +31,7 @@ pub async fn get_claims_by_ticker(client: Arc<Client>, ticker: &str) -> Result<V
 }
 
 #[tracing::instrument(skip(client, id))]
-pub async fn get_claim_by_id(client: Arc<Client>, id: Uuid) -> Result<Claim, Error> {
+pub async fn get_claim_by_id<T: GenericClient>(client: &T, id: Uuid) -> Result<Claim, Error> {
     trace!(%id, "Getting claim");
     client
         .query_one("SELECT * FROM claims WHERE id = $1", &[&id])
@@ -39,8 +40,8 @@ pub async fn get_claim_by_id(client: Arc<Client>, id: Uuid) -> Result<Claim, Err
 }
 
 #[tracing::instrument(skip(client, id, amount))]
-pub async fn update_claim_amount(
-    client: Arc<Client>,
+pub async fn update_claim_amount<T: GenericClient>(
+    client: &T,
     id: Uuid,
     amount: AmountSpec,
 ) -> Result<(), Error> {
@@ -56,7 +57,7 @@ pub async fn update_claim_amount(
 }
 
 #[tracing::instrument(skip(client, id))]
-pub async fn delete_claim_by_id(client: Arc<Client>, id: Uuid) -> Result<(), Error> {
+pub async fn delete_claim_by_id<T: GenericClient>(client: &T, id: Uuid) -> Result<(), Error> {
     trace!(%id, "Deleting claim");
     client
         .execute("DELETE FROM claims WHERE id = $1;", &[&id])
@@ -65,7 +66,7 @@ pub async fn delete_claim_by_id(client: Arc<Client>, id: Uuid) -> Result<(), Err
 }
 
 #[tracing::instrument(skip(client, claim))]
-pub async fn save_claim(client: Arc<Client>, claim: Claim) -> Result<(), Error> {
+pub async fn save_claim<T: GenericClient>(client: &T, claim: Claim) -> Result<(), Error> {
     trace!(id = %claim.id, "Saving claim");
     let (amount, unit) = split_amount_spec(claim.amount);
     client.execute("INSERT INTO claims (id, strategy, sub_strategy, ticker, amount, unit) VALUES ($1, $2, $3, $4, $5, $6);", &[

src/db/dependent_orders.rs

@@ -1,13 +1,12 @@
 use alpaca::orders::OrderIntent;
 use alpaca::OrderType;
 use anyhow::{anyhow, Result};
-use std::sync::Arc;
-use tokio_postgres::Client;
+use tokio_postgres::GenericClient;
 use tracing::trace;
 
 #[tracing::instrument(skip(client, id, dependent_order))]
-pub async fn save_dependent_order(
-    client: Arc<Client>,
+pub async fn save_dependent_order<T: GenericClient>(
+    client: &T,
     id: &str,
     dependent_order: OrderIntent,
 ) -> Result<()> {
@@ -43,7 +42,10 @@ pub async fn save_dependent_order(
 }
 
 #[tracing::instrument(skip(client, id))]
-pub async fn take_dependent_orders(client: Arc<Client>, id: &str) -> Result<Vec<OrderIntent>> {
+pub async fn take_dependent_orders<T: GenericClient>(
+    client: &T,
+    id: &str,
+) -> Result<Vec<OrderIntent>> {
     trace!(id, "Saving dependent order");
     client
         .query(

src/db/lots.rs

@@ -1,11 +1,10 @@
 use crate::types::Lot;
 use std::convert::TryInto;
-use std::sync::Arc;
-use tokio_postgres::{Client, Error};
+use tokio_postgres::{Error, GenericClient};
 use tracing::trace;
 
 #[tracing::instrument(skip(client))]
-pub async fn get_lots(client: Arc<Client>) -> Result<Vec<Lot>, Error> {
+pub async fn get_lots<T: GenericClient>(client: &T) -> Result<Vec<Lot>, Error> {
     trace!("Getting lots");
     client
         .query("SELECT * FROM lots", &[])
@@ -16,7 +15,10 @@ pub async fn get_lots(client: Arc<Client>) -> Result<Vec<Lot>, Error> {
 }
 
 #[tracing::instrument(skip(client, order_id))]
-pub async fn get_lots_by_order_id(client: Arc<Client>, order_id: &str) -> Result<Vec<Lot>, Error> {
+pub async fn get_lots_by_order_id<T: GenericClient>(
+    client: &T,
+    order_id: &str,
+) -> Result<Vec<Lot>, Error> {
     trace!(order_id, "Getting lots");
     client
         .query("SELECT * FROM lots WHERE order_id = $1", &[&order_id])
@@ -36,7 +38,7 @@ pub async fn get_lots_by_order_id(client: Arc<Client>, order_id: &str) -> Result
 }
 
 #[tracing::instrument(skip(client, lot))]
-pub async fn save_lot(client: Arc<Client>, lot: Lot) -> Result<(), Error> {
+pub async fn save_lot<T: GenericClient>(client: &T, lot: Lot) -> Result<(), Error> {
     trace!(id = %lot.id, "Saving lot");
     client.execute("INSERT INTO lots (id, order_id, ticker, fill_time, price, shares) VALUES ($1, $2, $3, $4, $5, $6);", &[
             &lot.id,

src/db/pending_orders.rs

@@ -1,12 +1,11 @@
 use crate::types::PendingOrder;
 use std::convert::TryInto;
-use std::sync::Arc;
-use tokio_postgres::{Client, Error};
+use tokio_postgres::{Error, GenericClient};
 use tracing::trace;
 
 #[tracing::instrument(skip(client, ticker))]
-pub async fn get_pending_order_amount_by_ticker(
-    client: Arc<Client>,
+pub async fn get_pending_order_amount_by_ticker<T: GenericClient>(
+    client: &T,
     ticker: &str,
 ) -> Result<Option<i32>, Error> {
     trace!(ticker, "Getting pending order amount");
@@ -21,7 +20,7 @@ pub async fn get_pending_order_amount_by_ticker(
 }
 
 #[tracing::instrument(skip(client))]
-pub async fn get_pending_orders(client: Arc<Client>) -> Result<Vec<PendingOrder>, Error> {
+pub async fn get_pending_orders<T: GenericClient>(client: &T) -> Result<Vec<PendingOrder>, Error> {
     trace!("Getting pending orders");
     client
         .query("SELECT * FROM pending_orders", &[])
@@ -32,8 +31,8 @@ pub async fn get_pending_orders(client: Arc<Client>) -> Result<Vec<PendingOrder>
 }
 
 #[tracing::instrument(skip(client, id))]
-pub async fn get_pending_order_by_id(
-    client: Arc<Client>,
+pub async fn get_pending_order_by_id<T: GenericClient>(
+    client: &T,
     id: &str,
 ) -> Result<Option<PendingOrder>, Error> {
     trace!(id, "Getting pending order");
@@ -45,8 +44,8 @@ pub async fn get_pending_order_by_id(
 }
 
 #[tracing::instrument(skip(client, id, qty))]
-pub async fn update_pending_order_qty(
-    client: Arc<Client>,
+pub async fn update_pending_order_qty<T: GenericClient>(
+    client: &T,
     id: &str,
     qty: i32,
 ) -> Result<(), Error> {
@@ -61,8 +60,8 @@ pub async fn update_pending_order_qty(
 }
 
 #[tracing::instrument(skip(client, pending_order))]
-pub async fn save_pending_order(
-    client: Arc<Client>,
+pub async fn save_pending_order<T: GenericClient>(
+    client: &T,
     pending_order: PendingOrder,
 ) -> Result<(), Error> {
     trace!(id = %pending_order.id, "Saving pending order");
@@ -71,7 +70,10 @@ pub async fn save_pending_order(
 }
 
 #[tracing::instrument(skip(client, id))]
-pub async fn delete_pending_order_by_id(client: Arc<Client>, id: &str) -> Result<(), Error> {
+pub async fn delete_pending_order_by_id<T: GenericClient>(
+    client: &T,
+    id: &str,
+) -> Result<(), Error> {
     trace!(id, "Deleting pending order");
     client
         .execute("DELETE FROM pending_orders WHERE id = $1", &[&id])

src/db/positions.rs

@@ -1,12 +1,11 @@
 use crate::types::{Owner, Position};
 use std::convert::TryInto;
-use std::sync::Arc;
-use tokio_postgres::{Client, Error};
+use tokio_postgres::{Error, GenericClient};
 use tracing::trace;
 
 #[tracing::instrument(skip(client, owner))]
-pub async fn get_positions_by_owner(
-    client: Arc<Client>,
+pub async fn get_positions_by_owner<T: GenericClient>(
+    client: &T,
     owner: Owner,
 ) -> Result<Vec<Position>, Error> {
     trace!("Getting positions");
@@ -27,8 +26,8 @@ pub async fn get_positions_by_owner(
 }
 
 #[tracing::instrument(skip(client, ticker))]
-pub async fn get_positions_by_ticker(
-    client: Arc<Client>,
+pub async fn get_positions_by_ticker<T: GenericClient>(
+    client: &T,
     ticker: &str,
 ) -> Result<Vec<Position>, Error> {
     trace!("Getting positions");
@@ -40,7 +39,7 @@ pub async fn get_positions_by_ticker(
 }
 
 #[tracing::instrument(skip(client))]
-pub async fn get_positions(client: Arc<Client>) -> Result<Vec<Position>, Error> {
+pub async fn get_positions<T: GenericClient>(client: &T) -> Result<Vec<Position>, Error> {
     trace!("Getting positions");
     client
         .query("SELECT * FROM allocations", &[])

src/db/scheduled_intents.rs

@@ -1,12 +1,12 @@
 use super::utils::{split_amount_spec, unite_amount_spec};
 use anyhow::Result;
-use std::sync::Arc;
-use tokio_postgres::Client;
+use tokio_postgres::GenericClient;
 use tracing::trace;
 use trading_base::{PositionIntent, TickerSpec};
 use uuid::Uuid;
 
-pub async fn get_scheduled_indents(client: Arc<Client>) -> Result<Vec<PositionIntent>> {
+#[tracing::instrument(skip(client))]
+pub async fn get_scheduled_indents<T: GenericClient>(client: &T) -> Result<Vec<PositionIntent>> {
     trace!("Getting scheduled intents");
     client
         .query("SELECT * FROM scheduled_intents", &[])
@@ -36,8 +36,8 @@ pub async fn get_scheduled_indents(client: Arc<Client>) -> Result<Vec<PositionIn
 }
 
 #[tracing::instrument(skip(client, scheduled_intent))]
-pub async fn save_scheduled_intent(
-    client: Arc<Client>,
+pub async fn save_scheduled_intent<T: GenericClient>(
+    client: &T,
     scheduled_intent: PositionIntent,
 ) -> Result<()> {
     trace!("Saving scheduled intent");
@@ -84,7 +84,7 @@ pub async fn save_scheduled_intent(
 }
 
 #[tracing::instrument(skip(client, id))]
-pub async fn delete_scheduled_intent(client: Arc<Client>, id: &Uuid) -> Result<()> {
+pub async fn delete_scheduled_intent<T: GenericClient>(client: &T, id: &Uuid) -> Result<()> {
     trace!(%id, "Deleting scheduled intent");
     client
         .execute("DELETE FROM scheduled_intents WHERE id = $1", &[&id])

src/order_manager/dependent_orders.rs

@@ -6,7 +6,7 @@ use tracing::debug;
 impl OrderManager {
     #[tracing::instrument(skip(self, id))]
     pub async fn trigger_dependent_orders(&mut self, id: &str) -> Result<()> {
-        let orders = db::take_dependent_orders(self.db_client.clone(), id)
+        let orders = db::take_dependent_orders(self.db_client.as_ref(), id)
             .await
             .context("Failed to take and delete dependent order")?;
         if !orders.is_empty() {

src/order_manager/input.rs

@@ -28,7 +28,7 @@ impl OrderManager {
             scheduled_intent = self.scheduler_receiver.recv() => {
                 debug!("Message received from scheduler");
                 let intent = scheduled_intent.ok_or_else(|| anyhow!("Channel closed"))?;
-                db::delete_scheduled_intent(self.db_client.clone(), &intent.id).await?;
+                db::delete_scheduled_intent(self.db_client.as_ref(), &intent.id).await?;
                 Ok(Input::PositionIntent(intent))
             }
         }