From 52b6d6ce4b2754d3fc1d47ff7d90f4663039ca95 Mon Sep 17 00:00:00 2001
From: akiroz <akiroz.vectis@gmail.com>
Date: Sat, 20 Jan 2024 10:17:18 +0900
Subject: [PATCH] Wrap passthru in mutex

---
 Cargo.lock    | 2 +-
 Cargo.toml    | 2 +-
 src/client.rs | 4 ++--
 src/remote.rs | 5 ++---
 4 files changed, 6 insertions(+), 7 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 5a94e69..cf4f909 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1315,7 +1315,7 @@ checksum = "525b4ec142c6b68a2d10f01f7bbf6755599ca3f81ea53b8431b7dd348f5fdb2d"
 
 [[package]]
 name = "zika"
-version = "3.3.0"
+version = "3.3.1"
 dependencies = [
  "base64",
  "bytes",
diff --git a/Cargo.toml b/Cargo.toml
index c5bc8fb..ad6c2af 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "zika"
-version = "3.3.0"
+version = "3.3.1"
 license = "MIT"
 description = "IP Tunneling over MQTT"
 repository = "https://github.com/akiroz/zika"
diff --git a/src/client.rs b/src/client.rs
index 2759e9a..3f6335e 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -24,7 +24,7 @@ pub struct Client {
     pub local_addr: Ipv4Addr,
     tunnels: Arc<Vec<Tunnel>>,
     pub remote: Arc<Mutex<remote::Remote>>, // Allow external mqtt ops
-    pub remote_passthru_recv: broadcast::Receiver<(String, Bytes)>,
+    pub remote_passthru_recv: Arc<Mutex<broadcast::Receiver<(String, Bytes)>>>,
 }
 
 struct Tunnel {
@@ -100,7 +100,7 @@ impl Client {
             local_addr,
             tunnels: Arc::new(tunnels),
             remote: Arc::new(Mutex::new(remote)),
-            remote_passthru_recv,
+            remote_passthru_recv: Arc::new(Mutex::new(remote_passthru_recv)),
         });
 
         let loop_client = client.clone();
diff --git a/src/remote.rs b/src/remote.rs
index a1c1c20..55a6967 100644
--- a/src/remote.rs
+++ b/src/remote.rs
@@ -1,12 +1,11 @@
-use bytes::Bytes;
 use log;
+use bytes::Bytes;
 use rumqttc::v5::{
     self as mqtt,
     mqttbytes::{v5::Packet, QoS},
 };
 use std::sync::Arc;
-use tokio::sync::Mutex;
-use tokio::{sync::mpsc, task};
+use tokio::{sync::{mpsc, Mutex}, task};
 
 use crate::lookup_pool::LookupPool;
 use rumqttc::v5::mqttbytes::v5::PublishProperties;