Skip to content

drop sub #495

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Aug 22, 2024
Merged

drop sub #495

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f8a7e8f
kns: sub for notes
nick1udwig Aug 21, 2024
fe50f31
add wip
nick1udwig Aug 21, 2024
9c20cf5
Format Rust code using rustfmt
github-actions[bot] Aug 21, 2024
e1106b4
eth: borrow fixes unsubscribe
barraguda Aug 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions kinode/packages/kns_indexer/kns_indexer/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -531,6 +531,12 @@ fn handle_log(
}
},
}

if !state.listening_newblocks && !pending_notes.is_empty() {
print_to_terminal(0, "subscribing to newHeads...");
listen_to_new_blocks_loop(); // sub_id: 3
state.listening_newblocks = true;
}
}
}
_log => {
Expand Down
7 changes: 4 additions & 3 deletions kinode/src/eth/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ type ResponseChannels = Arc<DashMap<u64, ProcessMessageSender>>;

#[derive(Debug)]
enum ActiveSub {
Local(JoinHandle<()>),
Local((tokio::sync::mpsc::Sender<bool>, JoinHandle<()>)),
Remote {
provider_node: String,
handle: JoinHandle<()>,
Expand All @@ -111,8 +111,9 @@ enum ActiveSub {
impl ActiveSub {
async fn close(&self, sub_id: u64, state: &ModuleState) {
match self {
ActiveSub::Local(handle) => {
handle.abort();
ActiveSub::Local((close_sender, _handle)) => {
close_sender.send(true).await.unwrap();
//handle.abort();
}
ActiveSub::Remote {
provider_node,
Expand Down
148 changes: 93 additions & 55 deletions kinode/src/eth/subscription.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,40 +70,53 @@ pub async fn create_new_subscription(
let send_to_loop = send_to_loop.clone();
let print_tx = print_tx.clone();
let active_subscriptions = active_subscriptions.clone();
let providers = providers.clone();
let (close_sender, close_receiver) = tokio::sync::mpsc::channel(1);
match maybe_raw_sub {
Ok(rx) => {
Ok((rx, chain_id)) => {
subs.insert(
sub_id,
// this is a local sub, as in, we connect to the rpc endpoint
ActiveSub::Local(tokio::spawn(async move {
// await the subscription error and kill it if so
let e = maintain_local_subscription(
&our,
sub_id,
rx,
&target,
&rsvp,
&send_to_loop,
&active_subscriptions,
)
.await;
verbose_print(
&print_tx,
&format!("eth: closed local subscription due to error {e:?}"),
)
.await;
kernel_message(
&our,
rand::random(),
target.clone(),
rsvp,
true,
None,
EthSubResult::Err(e),
&send_to_loop,
)
.await;
})),
ActiveSub::Local((
close_sender,
tokio::spawn(async move {
// await the subscription error and kill it if so
let r = maintain_local_subscription(
&our,
sub_id,
rx,
&target,
&rsvp,
&send_to_loop,
&active_subscriptions,
chain_id,
&providers,
close_receiver,
)
.await;
let Err(e) = r else {
return;
};
verbose_print(
&print_tx,
&format!(
"eth: closed local subscription due to error {e:?}"
),
)
.await;
kernel_message(
&our,
rand::random(),
target.clone(),
rsvp,
true,
None,
EthSubResult::Err(e),
&send_to_loop,
)
.await;
}),
)),
);
}
Err((provider_node, remote_sub_id)) => {
Expand Down Expand Up @@ -169,7 +182,7 @@ async fn build_subscription(
providers: &Providers,
response_channels: &ResponseChannels,
print_tx: &PrintSender,
) -> Result<Result<RawSubscription, (String, u64)>, EthError> {
) -> Result<Result<(RawSubscription, u64), (String, u64)>, EthError> {
let EthAction::SubscribeLogs {
chain_id,
kind,
Expand Down Expand Up @@ -244,7 +257,7 @@ async fn build_subscription(
)
.await;
}
return Ok(Ok(rx));
return Ok(Ok((rx, chain_id)));
}
Err(rpc_error) => {
verbose_print(
Expand Down Expand Up @@ -367,38 +380,63 @@ async fn maintain_local_subscription(
rsvp: &Option<Address>,
send_to_loop: &MessageSender,
active_subscriptions: &ActiveSubscriptions,
) -> EthSubError {
while let Ok(value) = rx.recv().await {
let result: SubscriptionResult = match serde_json::from_str(value.get()) {
Ok(res) => res,
Err(e) => {
return EthSubError {
id: sub_id,
error: e.to_string(),
chain_id: u64,
providers: &Providers,
mut close_receiver: tokio::sync::mpsc::Receiver<bool>,
) -> Result<(), EthSubError> {
loop {
tokio::select! {
_ = close_receiver.recv() => {
let alloy_sub_id = rx.local_id();
let alloy_sub_id = alloy_sub_id.clone().into();
let Some(chain_providers) = providers.get_mut(&chain_id) else {
return Ok(()); //?
};
for url in chain_providers.urls.iter() {
let Some(pubsub) = url.pubsub.as_ref() else {
continue;
};
let x = pubsub.unsubscribe(alloy_sub_id);
println!("we just tried unsubscribing unsubscribed: {:?}", x);
}
}
};
kernel_message(
our,
rand::random(),
target.clone(),
rsvp.clone(),
true,
None,
EthSubResult::Ok(EthSub { id: sub_id, result }),
&send_to_loop,
)
.await;
return Ok(());
},
value = rx.recv() => {
let Ok(value) = value else {
break;
};
let result: SubscriptionResult = match serde_json::from_str(value.get()) {
Ok(res) => res,
Err(e) => {
return Err(EthSubError {
id: sub_id,
error: e.to_string(),
});
}
};
kernel_message(
our,
rand::random(),
target.clone(),
rsvp.clone(),
true,
None,
EthSubResult::Ok(EthSub { id: sub_id, result }),
&send_to_loop,
)
.await;
},
}
}
active_subscriptions
.entry(target.clone())
.and_modify(|sub_map| {
sub_map.remove(&sub_id);
});
EthSubError {
Err(EthSubError {
id: sub_id,
error: "subscription closed unexpectedly".to_string(),
}
})
}

/// handle the subscription updates from a remote provider,
Expand Down