Skip to content

Commit

Permalink
Add delete message to ACL pub/sub tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@oteffahi
oteffahi committed Jul 24, 2024
1 parent 0c2ee09 commit 036ac96
Showing 1 changed file with 71 additions and 26 deletions.
97 changes: 71 additions & 26 deletions zenoh/tests/acl.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ mod test {
config,
config::{EndPoint, WhatAmI},
prelude::*,
sample::SampleKind,
Config, Session,
};
use zenoh_core::{zlock, ztimeout};
Expand Down Expand Up @@ -114,12 +115,20 @@ mod test {
{
let publisher = pub_session.declare_publisher(KEY_EXPR).await.unwrap();
let received_value = Arc::new(Mutex::new(String::new()));
let deleted = Arc::new(Mutex::new(false));

let temp_recv_value = received_value.clone();
let deleted_clone = deleted.clone();
let subscriber = sub_session
.declare_subscriber(KEY_EXPR)
.callback(move |sample| {
let mut temp_value = zlock!(temp_recv_value);
*temp_value = sample.payload().deserialize::<String>().unwrap();
if sample.kind() == SampleKind::Put {
let mut temp_value = zlock!(temp_recv_value);
*temp_value = sample.payload().deserialize::<String>().unwrap();
} else if sample.kind() == SampleKind::Delete {
let mut deleted = zlock!(deleted_clone);
*deleted = true;
}
})
.await
.unwrap();
Expand All @@ -128,6 +137,10 @@ mod test {
publisher.put(VALUE).await.unwrap();
tokio::time::sleep(SLEEP).await;
assert_ne!(*zlock!(received_value), VALUE);

publisher.delete().await.unwrap();
tokio::time::sleep(SLEEP).await;
assert_ne!(*zlock!(deleted), true);
ztimeout!(subscriber.undeclare()).unwrap();
}
close_sessions(sub_session, pub_session).await;
Expand Down Expand Up @@ -156,22 +169,32 @@ mod test {
{
let publisher = ztimeout!(pub_session.declare_publisher(KEY_EXPR)).unwrap();
let received_value = Arc::new(Mutex::new(String::new()));
let deleted = Arc::new(Mutex::new(false));

let temp_recv_value = received_value.clone();
let subscriber =
ztimeout!(sub_session
.declare_subscriber(KEY_EXPR)
.callback(move |sample| {
let deleted_clone = deleted.clone();
let subscriber = sub_session
.declare_subscriber(KEY_EXPR)
.callback(move |sample| {
if sample.kind() == SampleKind::Put {
let mut temp_value = zlock!(temp_recv_value);
*temp_value = sample.payload().deserialize::<String>().unwrap();
}))
} else if sample.kind() == SampleKind::Delete {
let mut deleted = zlock!(deleted_clone);
*deleted = true;
}
})
.await
.unwrap();

tokio::time::sleep(SLEEP).await;

ztimeout!(publisher.put(VALUE)).unwrap();
publisher.put(VALUE).await.unwrap();
tokio::time::sleep(SLEEP).await;

assert_eq!(*zlock!(received_value), VALUE);

publisher.delete().await.unwrap();
tokio::time::sleep(SLEEP).await;
assert_eq!(*zlock!(deleted), true);
ztimeout!(subscriber.undeclare()).unwrap();
}

Expand All @@ -196,6 +219,7 @@ mod test {
"flows": ["egress", "ingress"],
"messages": [
"put",
"delete",
"declare_subscriber"
],
"key_exprs": [
Expand Down Expand Up @@ -227,22 +251,32 @@ mod test {
{
let publisher = ztimeout!(pub_session.declare_publisher(KEY_EXPR)).unwrap();
let received_value = Arc::new(Mutex::new(String::new()));
let deleted = Arc::new(Mutex::new(false));

let temp_recv_value = received_value.clone();
let subscriber =
ztimeout!(sub_session
.declare_subscriber(KEY_EXPR)
.callback(move |sample| {
let deleted_clone = deleted.clone();
let subscriber = sub_session
.declare_subscriber(KEY_EXPR)
.callback(move |sample| {
if sample.kind() == SampleKind::Put {
let mut temp_value = zlock!(temp_recv_value);
*temp_value = sample.payload().deserialize::<String>().unwrap();
}))
} else if sample.kind() == SampleKind::Delete {
let mut deleted = zlock!(deleted_clone);
*deleted = true;
}
})
.await
.unwrap();

tokio::time::sleep(SLEEP).await;

ztimeout!(publisher.put(VALUE)).unwrap();
publisher.put(VALUE).await.unwrap();
tokio::time::sleep(SLEEP).await;

assert_ne!(*zlock!(received_value), VALUE);

publisher.delete().await.unwrap();
tokio::time::sleep(SLEEP).await;
assert_ne!(*zlock!(deleted), true);
ztimeout!(subscriber.undeclare()).unwrap();
}
close_sessions(sub_session, pub_session).await;
Expand All @@ -266,6 +300,7 @@ mod test {
"flows": ["egress", "ingress"],
"messages": [
"put",
"delete",
"declare_subscriber"
],
"key_exprs": [
Expand Down Expand Up @@ -297,22 +332,32 @@ mod test {
{
let publisher = ztimeout!(pub_session.declare_publisher(KEY_EXPR)).unwrap();
let received_value = Arc::new(Mutex::new(String::new()));
let deleted = Arc::new(Mutex::new(false));

let temp_recv_value = received_value.clone();
let subscriber =
ztimeout!(sub_session
.declare_subscriber(KEY_EXPR)
.callback(move |sample| {
let deleted_clone = deleted.clone();
let subscriber = sub_session
.declare_subscriber(KEY_EXPR)
.callback(move |sample| {
if sample.kind() == SampleKind::Put {
let mut temp_value = zlock!(temp_recv_value);
*temp_value = sample.payload().deserialize::<String>().unwrap();
}))
} else if sample.kind() == SampleKind::Delete {
let mut deleted = zlock!(deleted_clone);
*deleted = true;
}
})
.await
.unwrap();

tokio::time::sleep(SLEEP).await;

ztimeout!(publisher.put(VALUE)).unwrap();
publisher.put(VALUE).await.unwrap();
tokio::time::sleep(SLEEP).await;

assert_eq!(*zlock!(received_value), VALUE);

publisher.delete().await.unwrap();
tokio::time::sleep(SLEEP).await;
assert_eq!(*zlock!(deleted), true);
ztimeout!(subscriber.undeclare()).unwrap();
}
close_sessions(sub_session, pub_session).await;
Expand Down

0 comments on commit 036ac96

Please sign in to comment.