Skip to content

Commit 527e6ec

Browse files
committed
use unbounded_send when sending pub/sub message from network handler to client
- improved logs
1 parent c77bf5f commit 527e6ec

File tree

1 file changed

+38
-20
lines changed

1 file changed

+38
-20
lines changed

src/network/network_handler.rs

+38-20
Original file line numberDiff line numberDiff line change
@@ -223,13 +223,13 @@ impl NetworkHandler {
223223
}
224224
Status::Subscribed => {
225225
for command in &msg.commands {
226-
if let "UNSUBSCRIBE" | "PUNSUBSCRIBE" | "SUNSUBSCRIBE" = command.name {
227-
let subscription_type = match command.name {
228-
"UNSUBSCRIBE" => SubscriptionType::Channel,
229-
"PUNSUBSCRIBE" => SubscriptionType::Pattern,
230-
"SUNSUBSCRIBE" => SubscriptionType::ShardChannel,
231-
_ => unreachable!(),
232-
};
226+
let subscription_type = match command.name {
227+
"UNSUBSCRIBE" => Some(SubscriptionType::Channel),
228+
"PUNSUBSCRIBE" => Some(SubscriptionType::Pattern),
229+
"SUNSUBSCRIBE" => Some(SubscriptionType::ShardChannel),
230+
_ => None,
231+
};
232+
if let Some(subscription_type) = subscription_type {
233233
self.pending_unsubscriptions.push_back(
234234
command
235235
.args
@@ -568,16 +568,34 @@ impl NetworkHandler {
568568
| RefPubSubMessage::SMessage(channel_or_pattern, _) => {
569569
match self.subscriptions.get_mut(channel_or_pattern) {
570570
Some((_subscription_type, pub_sub_sender)) => {
571-
if let Err(e) = pub_sub_sender.send(value).await {
572-
warn!(
573-
"[{}] Cannot send pub/sub message to caller: {e}",
574-
self.tag
575-
);
571+
if let Err(e) = pub_sub_sender.unbounded_send(value) {
572+
let error_desc = e.to_string();
573+
if let Ok(ref_value) = &e.into_inner() {
574+
if let Some(pub_sub_message) =
575+
RefPubSubMessage::from_resp(ref_value)
576+
{
577+
if let RefPubSubMessage::Message(
578+
channel_or_pattern,
579+
_,
580+
)
581+
| RefPubSubMessage::SMessage(
582+
channel_or_pattern,
583+
_,
584+
) = pub_sub_message
585+
{
586+
warn!(
587+
"[{}] Cannot send pub/sub message to caller from channel `{}`: {error_desc}",
588+
self.tag,
589+
String::from_utf8_lossy(channel_or_pattern)
590+
);
591+
}
592+
}
593+
}
576594
}
577595
}
578596
None => {
579597
error!(
580-
"[{}] Unexpected message on channel '{:?}'",
598+
"[{}] Unexpected message on channel `{}`",
581599
self.tag,
582600
String::from_utf8_lossy(channel_or_pattern)
583601
);
@@ -600,7 +618,7 @@ impl NetworkHandler {
600618
{
601619
return Some(Err(Error::Client(
602620
format!(
603-
"There is already a subscription on channel '{}'",
621+
"There is already a subscription on channel `{}`",
604622
String::from_utf8_lossy(channel_or_pattern)
605623
)
606624
.to_string(),
@@ -612,14 +630,14 @@ impl NetworkHandler {
612630
}
613631
} else {
614632
error!(
615-
"[{}] Unexpected subscription confirmation on channel '{}'",
633+
"[{}] Unexpected subscription confirmation on channel `{}`",
616634
self.tag,
617635
String::from_utf8_lossy(channel_or_pattern)
618636
);
619637
}
620638
} else {
621639
error!(
622-
"[{}] Cannot find pending subscription for channel '{}'",
640+
"[{}] Cannot find pending subscription for channel `{}`",
623641
self.tag,
624642
String::from_utf8_lossy(channel_or_pattern)
625643
);
@@ -634,7 +652,7 @@ impl NetworkHandler {
634652
if remaining.len() > 1 {
635653
if remaining.remove(channel_or_pattern).is_none() {
636654
error!(
637-
"[{}] Cannot find channel or pattern to remove: {}",
655+
"[{}] Cannot find channel or pattern to remove: `{}`",
638656
self.tag,
639657
String::from_utf8_lossy(channel_or_pattern)
640658
);
@@ -645,15 +663,15 @@ impl NetworkHandler {
645663
let Some(mut remaining) = self.pending_unsubscriptions.pop_front()
646664
else {
647665
error!(
648-
"[{}] Cannot find channel or pattern to remove: {}",
666+
"[{}] Cannot find channel or pattern to remove: `{}`",
649667
self.tag,
650668
String::from_utf8_lossy(channel_or_pattern)
651669
);
652670
return None;
653671
};
654672
if remaining.remove(channel_or_pattern).is_none() {
655673
error!(
656-
"[{}] Cannot find channel or pattern to remove: {}",
674+
"[{}] Cannot find channel or pattern to remove: `{}`",
657675
self.tag,
658676
String::from_utf8_lossy(channel_or_pattern)
659677
);
@@ -677,7 +695,7 @@ impl NetworkHandler {
677695
}
678696
None => {
679697
error!(
680-
"[{}] Unexpected message on channel '{}' for pattern '{}'",
698+
"[{}] Unexpected message on channel `{}` for pattern `{}`",
681699
self.tag,
682700
String::from_utf8_lossy(channel),
683701
String::from_utf8_lossy(pattern)

0 commit comments

Comments
 (0)