andrey/messagebus
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Bump version

Browse Source
This commit is contained in:
Andrey Tkachenko 2021-09-28 16:42:09 +04:00
parent 535c39c9dc
commit 7034e8b87b
3 changed files with 13 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.toml
View File

@ -1,6 +1,6 @@
[package]
name = "messagebus"
version = "0.9.7"
version = "0.9.8"
authors = ["Andrey Tkachenko <andrey@aidev.ru>"]
repository = "https://github.com/andreytkachenko/messagebus.git"
keywords = ["futures", "async", "tokio", "message", "bus"]

4
crates/remote/Cargo.toml
View File

@ -15,7 +15,7 @@ edition = "2018"
[dependencies]
thiserror = "1.0"
messagebus = { path="../../" }
messagebus = "0.9"
tokio = { version = "1", features = ["parking_lot", "rt-multi-thread", "sync", "time", "io-util"] }
parking_lot = "0.11"
quinn = "0.7"
@ -31,7 +31,7 @@ bytes = "1.1.0"
quinn-proto = "0.7.3"
rustls = "0.19.1"
redis = {version = "0.21.2", features = ["aio", "tokio-comp"]}
bitflags = "1.3.2"
bitflags = "1.2.1"
serde_json = "1.0.68"
log = "0.4.14"

18
crates/remote/src/relays/quic/mod.rs
View File

@ -41,8 +41,8 @@ enum RecvDo {
pub struct QuicRelay<B> {
base: Mutex<Option<B>>,
self_id: Arc<AtomicU64>,
outgoing_table: MessageTable,
in_table: MessageTable,
out_table: MessageTable,
item_sender: UnboundedSender<Option<ProtocolItem>>,
item_receiver: Mutex<Option<UnboundedReceiver<Option<ProtocolItem>>>>,
event_sender: UnboundedSender<RecvDo>,
@ -50,14 +50,15 @@ pub struct QuicRelay<B> {
}
impl QuicRelay<QuicClientEndpoint> {
pub fn new(cert: &str, addr: SocketAddr, host: String, table: Vec<(TypeTag, TypeTag, TypeTag)>) -> Result<Self, crate::error::Error> {
pub fn new(cert: &str, addr: SocketAddr, host: String, table: (Vec<(TypeTag, TypeTag, TypeTag)>, Vec<(TypeTag, TypeTag, TypeTag)>)) -> Result<Self, crate::error::Error> {
let (item_sender, item_receiver) = mpsc::unbounded_channel();
let (event_sender, event_receiver) = mpsc::unbounded_channel();
Ok(QuicRelay {
base: Mutex::new(Some(QuicClientEndpoint::new(cert, addr, host)?)),
self_id: Arc::new(AtomicU64::new(0)),
outgoing_table: MessageTable::from(table),
in_table: MessageTable::from(table.0),
out_table: MessageTable::from(table.1),
item_sender,
item_receiver: Mutex::new(Some(item_receiver)),
event_sender,
@ -67,14 +68,15 @@ impl QuicRelay<QuicClientEndpoint> {
}
impl QuicRelay<QuicServerEndpoint> {
pub fn new(key_path: &str, cert_path: &str, addr: SocketAddr, table: Vec<(TypeTag, TypeTag, TypeTag)>) -> Result<Self, crate::error::Error> {
pub fn new(key_path: &str, cert_path: &str, addr: SocketAddr, table: (Vec<(TypeTag, TypeTag, TypeTag)>, Vec<(TypeTag, TypeTag, TypeTag)>)) -> Result<Self, crate::error::Error> {
let (item_sender, item_receiver) = mpsc::unbounded_channel();
let (event_sender, event_receiver) = mpsc::unbounded_channel();
Ok(QuicRelay {
base: Mutex::new(Some(QuicServerEndpoint::new(key_path, cert_path, &addr )?)),
self_id: Arc::new(AtomicU64::new(0)),
outgoing_table: MessageTable::from(table),
in_table: MessageTable::from(table.0),
out_table: MessageTable::from(table.1),
item_sender,
item_receiver: Mutex::new(Some(item_receiver)),
event_sender,
@ -87,11 +89,11 @@ impl<B> TypeTagAccept for QuicRelay<B>
where B: Stream<Item = Connecting> + Send + 'static
{
fn accept(&self, msg: &TypeTag, resp: Option<&TypeTag>, err: Option<&TypeTag>) -> bool {
self.outgoing_table.accept(msg, resp, err)
self.in_table.accept(msg, resp, err)
}
fn iter_types(&self, cb: &mut dyn FnMut(&TypeTag, &TypeTag, &TypeTag) -> bool) {
let iter = self.outgoing_table.iter_types();
let iter = self.in_table.iter_types();
for (m, r, e) in iter {
if cb(m, r, e) {