Upgrade dependencies

rmqtt-bin/Cargo.toml

@@ -57,5 +57,5 @@ rmqtt-sys-topic = { }
 #rmqtt-plugin-template = { }
 
 [build-dependencies]
-toml = "0.5"
+toml = "0.7"
 

rmqtt-macros/Cargo.toml

@@ -11,4 +11,4 @@ proc-macro = true
 [dependencies]
 proc-macro2 = "1.0"
 quote = "1.0"
-syn = "1.0"
+syn = "2.0"

rmqtt-plugins/rmqtt-http-api/src/api.rs

@@ -7,10 +7,11 @@ use salvo::hyper::server::conn::AddrIncoming;
 use salvo::prelude::*;
 
 use rmqtt::{
-    anyhow, base64, bytes, chrono, futures, log,
+    anyhow, bytes, chrono, futures, log,
     serde_json::{self, json},
     tokio::sync::oneshot,
     HashMap,
+    base64::{Engine as _, engine::general_purpose},
 };
 use rmqtt::{
     broker::types::NodeId,
@@ -707,7 +708,7 @@ async fn _publish(
     let payload = if encoding == "plain" {
         bytes::Bytes::from(params.payload)
     } else if encoding == "base64" {
-        bytes::Bytes::from(base64::decode(params.payload).map_err(anyhow::Error::new)?)
+        bytes::Bytes::from(general_purpose::STANDARD.decode(params.payload).map_err(anyhow::Error::new)?)
     } else {
         return Err(MqttError::Msg("encoding error, currently only plain and base64 are supported".into()));
     };

rmqtt-plugins/rmqtt-sys-topic/src/lib.rs

@@ -5,13 +5,13 @@ extern crate serde;
 use config::PluginConfig;
 use rmqtt::{
     async_trait::async_trait,
-    base64,
     bytes::Bytes,
     chrono, log,
     serde_json::{self, json},
     tokio::spawn,
     tokio::sync::RwLock,
     tokio::time::sleep,
+    base64::{Engine as _, engine::general_purpose},
 };
 use rmqtt::{
     broker::hook::{Handler, HookResult, Parameter, Register, ReturnType, Type},
@@ -281,7 +281,7 @@ impl Handler for SystemTopicHandler {
                     "qos": publish.qos().value(),
                     "topic": publish.topic(),
                     "packet_id": publish.packet_id(),
-                    "payload": base64::encode(publish.payload()),
+                    "payload": general_purpose::STANDARD.encode(publish.payload()),
                     "reason": reason.to_string(),
                     "pts": publish.create_time(),
                     "ts": now.timestamp_millis(),

rmqtt-plugins/rmqtt-web-hook/src/lib.rs

@@ -18,7 +18,7 @@ use config::PluginConfig;
 use rmqtt::{
     anyhow::anyhow,
     async_trait::async_trait,
-    base64,
+    base64::{Engine as _, engine::general_purpose},
     bytestring::ByteString,
     chrono, futures, lazy_static, log,
     once_cell::sync::OnceCell,
@@ -623,7 +623,7 @@ impl Handler for WebHookHandler {
                     "qos": publish.qos().value(),
                     "topic": topic,
                     "packet_id": publish.packet_id(),
-                    "payload": base64::encode(publish.payload()),
+                    "payload": general_purpose::STANDARD.encode(publish.payload()),
                     "ts": publish.create_time(),
                     "time": now_time
                 });
@@ -642,7 +642,7 @@ impl Handler for WebHookHandler {
                         "qos": publish.qos().value(),
                         "topic": topic,
                         "packet_id": publish.packet_id(),
-                        "payload": base64::encode(publish.payload()),
+                        "payload": general_purpose::STANDARD.encode(publish.payload()),
                         "pts": publish.create_time(),
                         "ts": now.timestamp_millis(),
                         "time": now_time
@@ -664,7 +664,7 @@ impl Handler for WebHookHandler {
                         "qos": publish.qos().value(),
                         "topic": topic,
                         "packet_id": publish.packet_id(),
-                        "payload": base64::encode(publish.payload()),
+                        "payload": general_purpose::STANDARD.encode(publish.payload()),
                         "pts": publish.create_time(),
                         "ts": now.timestamp_millis(),
                         "time": now_time
@@ -685,7 +685,7 @@ impl Handler for WebHookHandler {
                         "qos": publish.qos().value(),
                         "topic": publish.topic(),
                         "packet_id": publish.packet_id(),
-                        "payload": base64::encode(publish.payload()),
+                        "payload": general_purpose::STANDARD.encode(publish.payload()),
                         "reason": reason.to_string(),
                         "pts": publish.create_time(),
                         "ts": now.timestamp_millis(),

rmqtt/Cargo.toml

@@ -27,10 +27,10 @@ ntex-mqtt = { git = "https://github.com/rmqtt/ntex-mqtt", branch = "0.6.18" }
 #ntex-mqtt = { path = "../../ntex-mqtt" }
 futures = "0.3"
 tokio = { version = "1", features = ["sync", "time", "macros", "rt", "rt-multi-thread", "fs"] }
-socket2 = { version = "0.4", features = ["all"] }
+socket2 = { version = "0.5", features = ["all"] }
 tokio-stream = { version = "0.1", features = ["net"] }
-tonic = "0.8"
-prost = "0.11"
+tonic = "0.10"
+prost = "0.12"
 once_cell = "1.10"
 dashmap = "5.4"
 ahash = "0.8"
@@ -47,30 +47,29 @@ chrono = { version = "0.4", default-features = false, features = ["clock"] }
 uuid = { version = "1.1", features = ["v4"] }
 rand = "0.8"
 crossbeam = "0.8"
-governor = "0.3"
-config = { version = "0.11.0", default-features = false, features = ["toml"] }
+governor = "0.6"
+config = { version = "0.11", default-features = false, features = ["toml"] }
 log = { version = "0.4", features = ["std"] }
 slog = "2.7"
 slog-term = "2.9"
 slog-async = "2.7"
 slog-stdlog = "4.1"
 slog-scope = "4.4"
-base64 = "0.13"
+base64 = "0.21"
 bincode = "1.3"
 url = { version = "2.2", default-features = false }
-systemstat = "0.1"
-itertools = "0.10"
+systemstat = "0.2"
+itertools = "0.11"
 reqwest = { version = "0.11", default-features = false, features = ["rustls-tls", "json"] }
 rust-box = { version = "0.7.1", features = ["task-exec-queue", "task-exec-queue-rate", "std-ext", "dequemap"] }
 structopt = "0.3"
-tokio-tungstenite = "0.18"
+tokio-tungstenite = "0.20"
 pin-project-lite = "0.2.5"
 tokio-cron-scheduler = "0.9.4"
 update_rate = "2.0"
 bitflags = "2.3.3"
-time = "=0.3.20"
 
 [build-dependencies]
-tonic-build = "0.8"
-toml = "0.5"
+tonic-build = "0.10"
+toml = "0.7"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }

rmqtt/src/broker/queue.rs

@@ -15,11 +15,12 @@ use governor::{
     prelude::StreamRateLimitExt,
     state::{InMemoryState, NotKeyed},
     Quota, RateLimiter, RatelimitedStream,
+    middleware::NoOpMiddleware
 };
 
 type DirectLimiter = RateLimiter<NotKeyed, InMemoryState, DefaultClock>;
 
-pub type Receiver<'a, T> = RatelimitedStream<'a, ReceiverStream<T>, InMemoryState, DefaultClock>;
+pub type Receiver<'a, T> = RatelimitedStream<'a, ReceiverStream<T>, InMemoryState, DefaultClock, NoOpMiddleware>;
 
 pub enum Policy {
     //Discard current value

rmqtt/src/broker/types.rs

@@ -9,6 +9,7 @@ use std::rc::Rc;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+use base64::{Engine as _, engine::general_purpose};
 use bitflags::bitflags;
 use bytestring::ByteString;
 use itertools::Itertools;
@@ -904,15 +905,15 @@ impl<'a> LastWill<'a> {
                     "qos": lw.qos.value(),
                     "retain": lw.retain,
                     "topic": lw.topic,
-                    "message": base64::encode(lw.message.as_ref()),
+                    "message": general_purpose::STANDARD.encode(lw.message.as_ref()),
                 })
             }
             LastWill::V5(lw) => {
                 json!({
                     "qos": lw.qos.value(),
                     "retain": lw.retain,
                     "topic": lw.topic,
-                    "message": base64::encode(lw.message.as_ref()),
+                    "message": general_purpose::STANDARD.encode(lw.message.as_ref()),
 
                     "will_delay_interval_sec": lw.will_delay_interval_sec,
                     "correlation_data": lw.correlation_data,