lukechilds
/
itchysats
mirror of https://github.com/lukechilds/itchysats.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge #414 

414: Initialize Cfd actors outside of Rocket r=klochowicz a=klochowicz

Allows us to use `?` instead of unwrapping.

Co-authored-by: Mariusz Klochowicz <mariusz@klochowicz.com>
feature/actor-custom-derive
bors[bot] 3 years ago
committed by GitHub
parent
db28c23bc2 8e2f3cfc95
commit
d0f8d932e2
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 104 additions and 147 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 76
      daemon/src/maker.rs
  2. 71
      daemon/src/taker.rs

76
daemon/src/maker.rs
View File

@ -15,7 +15,6 @@ use daemon::{
oracle, wallet_sync,
};
use futures::Future;
use rocket::fairing::AdHoc;
use sqlx::sqlite::SqliteConnectOptions;
use sqlx::SqlitePool;
use std::collections::HashMap;
@ -176,33 +175,15 @@ async fn main() -> Result<()> {
)
.await?;
rocket::custom(figment)
.manage(wallet_feed_receiver)
.manage(auth_password)
.manage(quote_updates)
.manage(bitcoin_network)
.attach(AdHoc::try_on_ignite("SQL migrations", {
let db = db.clone();
move |rocket| async move {
match db::run_migrations(&db).await {
Ok(_) => Ok(rocket),
Err(_) => Err(rocket),
}
}
}))
.attach(AdHoc::try_on_ignite("Create actors", {
let db = db.clone();
db::run_migrations(&db)
.await
.context("Db migrations failed")?;
move |rocket| async move {
let mut conn = db.acquire().await.unwrap();
// Create actors
let mut conn = db.acquire().await?;
housekeeping::transition_non_continue_cfds_to_setup_failed(&mut conn)
.await
.unwrap();
housekeeping::rebroadcast_transactions(&mut conn, &wallet)
.await
.unwrap();
housekeeping::transition_non_continue_cfds_to_setup_failed(&mut conn).await?;
housekeeping::rebroadcast_transactions(&mut conn, &wallet).await?;
let ActorSystem {
cfd_actor_addr,
@ -210,7 +191,7 @@ async fn main() -> Result<()> {
order_feed_receiver,
update_cfd_feed_receiver,
} = ActorSystem::new(
db,
db.clone(),
wallet.clone(),
oracle,
|cfds, channel| oracle::Actor::new(cfds, channel),
@ -223,23 +204,23 @@ async fn main() -> Result<()> {
|channel0, channel1| maker_inc_connections::Actor::new(channel0, channel1),
listener,
)
.await;
.await?;
tokio::spawn(wallet_sync::new(wallet, wallet_feed_sender));
let cfd_action_channel =
MessageChannel::<maker_cfd::CfdAction>::clone_channel(&cfd_actor_addr);
let new_order_channel =
MessageChannel::<maker_cfd::NewOrder>::clone_channel(&cfd_actor_addr);
let cfd_action_channel = MessageChannel::<maker_cfd::CfdAction>::clone_channel(&cfd_actor_addr);
let new_order_channel = MessageChannel::<maker_cfd::NewOrder>::clone_channel(&cfd_actor_addr);
Ok(rocket
rocket::custom(figment)
.manage(order_feed_receiver)
.manage(update_cfd_feed_receiver)
.manage(cfd_action_channel)
.manage(new_order_channel)
.manage(cfd_feed_receiver))
}
}))
.manage(cfd_feed_receiver)
.manage(wallet_feed_receiver)
.manage(auth_password)
.manage(quote_updates)
.manage(bitcoin_network)
.mount(
"/api",
rocket::routes![
@ -295,13 +276,13 @@ where
Box<dyn MessageChannel<FromTaker>>,
) -> T,
listener: TcpListener,
) -> Self
) -> Result<Self>
where
F: Future<Output = Result<M>>,
{
let mut conn = db.acquire().await.unwrap();
let mut conn = db.acquire().await?;
let cfds = load_all_cfds(&mut conn).await.unwrap();
let cfds = load_all_cfds(&mut conn).await?;
let (cfd_feed_sender, cfd_feed_receiver) = watch::channel(cfds.clone());
let (order_feed_sender, order_feed_receiver) = watch::channel::<Option<Order>>(None);
@ -334,20 +315,17 @@ where
tokio::spawn(
monitor_ctx
.notify_interval(Duration::from_secs(20), || monitor::Sync)
.unwrap(),
.map_err(|e| anyhow::anyhow!(e))?,
);
tokio::spawn(
monitor_ctx.run(
monitor_constructor(Box::new(cfd_actor_addr.clone()), cfds.clone())
.await
.unwrap(),
),
monitor_ctx
.run(monitor_constructor(Box::new(cfd_actor_addr.clone()), cfds.clone()).await?),
);
tokio::spawn(
oracle_ctx
.notify_interval(Duration::from_secs(5), || oracle::Sync)
.unwrap(),
.map_err(|e| anyhow::anyhow!(e))?,
);
let fan_out_actor = fan_out::Actor::new(&[&cfd_actor_addr, &monitor_addr])
.create(None)
@ -355,7 +333,7 @@ where
tokio::spawn(oracle_ctx.run(oracle_constructor(cfds, Box::new(fan_out_actor))));
oracle_addr.do_send_async(oracle::Sync).await.unwrap();
oracle_addr.do_send_async(oracle::Sync).await?;
let listener_stream = futures::stream::poll_fn(move |ctx| {
let message = match futures::ready!(listener.poll_accept(ctx)) {
@ -370,11 +348,11 @@ where
tokio::spawn(inc_conn_addr.attach_stream(listener_stream));
Self {
Ok(Self {
cfd_actor_addr,
cfd_feed_receiver,
order_feed_receiver,
update_cfd_feed_receiver,
}
})
}
}

71
daemon/src/taker.rs
View File

@ -12,7 +12,6 @@ use daemon::{
bitmex_price_feed, fan_out, housekeeping, logger, monitor, oracle, taker_cfd, wallet_sync, wire,
};
use futures::{Future, Stream};
use rocket::fairing::AdHoc;
use sqlx::sqlite::SqliteConnectOptions;
use sqlx::SqlitePool;
use std::collections::HashMap;
@ -153,32 +152,15 @@ async fn main() -> Result<()> {
)
.await?;
rocket::custom(figment)
.manage(wallet_feed_receiver)
.manage(quote_updates)
.manage(bitcoin_network)
.attach(AdHoc::try_on_ignite("SQL migrations", {
let db = db.clone();
move |rocket| async move {
match db::run_migrations(&db).await {
Ok(_) => Ok(rocket),
Err(_) => Err(rocket),
}
}
}))
.attach(AdHoc::try_on_ignite("Create actors", {
let db = db.clone();
db::run_migrations(&db)
.await
.context("Db migrations failed")?;
move |rocket| async move {
let mut conn = db.acquire().await.unwrap();
// Create actors
let mut conn = db.acquire().await?;
housekeeping::transition_non_continue_cfds_to_setup_failed(&mut conn)
.await
.unwrap();
housekeeping::rebroadcast_transactions(&mut conn, &wallet)
.await
.unwrap();
housekeeping::transition_non_continue_cfds_to_setup_failed(&mut conn).await?;
housekeeping::rebroadcast_transactions(&mut conn, &wallet).await?;
let connection::Actor {
send_to_maker,
@ -191,7 +173,7 @@ async fn main() -> Result<()> {
order_feed_receiver,
update_cfd_feed_receiver,
} = ActorSystem::new(
db,
db.clone(),
wallet.clone(),
oracle,
send_to_maker,
@ -204,21 +186,21 @@ async fn main() -> Result<()> {
}
},
)
.await;
.await?;
tokio::spawn(wallet_sync::new(wallet, wallet_feed_sender));
let take_offer_channel =
MessageChannel::<taker_cfd::TakeOffer>::clone_channel(&cfd_actor_addr);
let cfd_action_channel =
MessageChannel::<taker_cfd::CfdAction>::clone_channel(&cfd_actor_addr);
Ok(rocket
let take_offer_channel = MessageChannel::<taker_cfd::TakeOffer>::clone_channel(&cfd_actor_addr);
let cfd_action_channel = MessageChannel::<taker_cfd::CfdAction>::clone_channel(&cfd_actor_addr);
rocket::custom(figment)
.manage(order_feed_receiver)
.manage(update_cfd_feed_receiver)
.manage(take_offer_channel)
.manage(cfd_action_channel)
.manage(cfd_feed_receiver))
}
}))
.manage(cfd_feed_receiver)
.manage(wallet_feed_receiver)
.manage(quote_updates)
.manage(bitcoin_network)
.mount(
"/api",
rocket::routes![
@ -267,13 +249,13 @@ where
read_from_maker: Box<dyn Stream<Item = taker_cfd::MakerStreamMessage> + Unpin + Send>,
oracle_constructor: impl Fn(Vec<Cfd>, Box<dyn StrongMessageChannel<Attestation>>) -> O,
monitor_constructor: impl Fn(Box<dyn StrongMessageChannel<monitor::Event>>, Vec<Cfd>) -> F,
) -> Self
) -> Result<Self>
where
F: Future<Output = Result<M>>,
{
let mut conn = db.acquire().await.unwrap();
let mut conn = db.acquire().await?;
let cfds = load_all_cfds(&mut conn).await.unwrap();
let cfds = load_all_cfds(&mut conn).await?;
let (cfd_feed_sender, cfd_feed_receiver) = watch::channel(cfds.clone());
let (order_feed_sender, order_feed_receiver) = watch::channel::<Option<Order>>(None);
@ -302,14 +284,11 @@ where
tokio::spawn(
monitor_ctx
.notify_interval(Duration::from_secs(20), || monitor::Sync)
.unwrap(),
.map_err(|e| anyhow::anyhow!(e))?,
);
tokio::spawn(
monitor_ctx.run(
monitor_constructor(Box::new(cfd_actor_addr.clone()), cfds.clone())
.await
.unwrap(),
),
monitor_ctx
.run(monitor_constructor(Box::new(cfd_actor_addr.clone()), cfds.clone()).await?),
);
tokio::spawn(
@ -323,11 +302,11 @@ where
tokio::spawn(oracle_ctx.run(oracle_constructor(cfds, Box::new(fan_out_actor))));
Self {
Ok(Self {
cfd_actor_addr,
cfd_feed_receiver,
order_feed_receiver,
update_cfd_feed_receiver,
}
})
}
}

Write Preview
Loading…
Cancel
Save