mod tests_utils;
use crate::tests_utils::checker::State;
use crate::tests_utils::test_channel::ChannelState;
use crate::tests_utils::test_runtime::block_on;
use crate::tests_utils::types::TestMutableHelper;
use rx_rust::disposable::Disposable;
use rx_rust::disposable::subscription::Subscription;
use rx_rust::operators::creating::empty::Empty;
use rx_rust::operators::creating::throw::Throw;
use rx_rust::subject::behavior_subject::BehaviorSubject;
use rx_rust::utils::types::{Mutable, Shared};
use rx_rust::{
observable::{Observable, observable_ext::ObservableExt},
observer::{Observer, Termination, boxed_observer::BoxedObserver},
operators::{creating::create::Create, transforming::window::Window},
subject::{publish_subject::PublishSubject, subject_observable::SubjectObservable},
};
use rx_rust::{safe_lock, safe_lock_vec};
use std::convert::Infallible;
use tests_utils::{checker::Checker, test_channel::test_channel, test_struct::TestStruct};
#[test]
fn test_completed_from_source() {
let (mut sender, observable, channel_checker) = test_channel::<'_, _, Infallible>();
let (mut boundary_sender, boundary_observable, boundary_channel_checker) = test_channel();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = observable.window(boundary_observable);
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(111);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(222);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(333);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_termination(Termination::Completed);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Completed);
assert_eq!(channel_checker.state(), ChannelState::Completed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Unsubscribed);
}
#[test]
fn test_completed_from_boundary() {
let (mut sender, observable, channel_checker) = test_channel::<'_, _, Infallible>();
let (mut boundary_sender, boundary_observable, boundary_channel_checker) = test_channel();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = observable.window(boundary_observable);
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(111);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(222);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(333);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_termination(Termination::Completed);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Completed);
assert_eq!(channel_checker.state(), ChannelState::Unsubscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Completed);
}
#[test]
fn test_completed_source_and_boundary_are_same() {
let mut subject = PublishSubject::<'_, _, Infallible>::default();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = subject.clone().window(subject.clone());
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
subject.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [()]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
subject.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [()]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), [()]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
subject.on_termination(Termination::Completed);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [()]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), [()]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Completed);
}
#[test]
fn test_error_from_source() {
let (mut sender, observable, channel_checker) = test_channel();
let (mut boundary_sender, boundary_observable, boundary_channel_checker) = test_channel();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = observable.window(boundary_observable);
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(111);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(222);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(333);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_termination(Termination::Error("error"));
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Error("error"));
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Error("error"));
assert_eq!(channel_checker.state(), ChannelState::Error("error"));
assert_eq!(boundary_channel_checker.state(), ChannelState::Unsubscribed);
}
#[test]
fn test_error_from_boundary() {
let (mut sender, observable, channel_checker) = test_channel();
let (mut boundary_sender, boundary_observable, boundary_channel_checker) = test_channel();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = observable.window(boundary_observable);
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(111);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(222);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(333);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_termination(Termination::Error("error"));
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Error("error"));
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Error("error"));
assert_eq!(channel_checker.state(), ChannelState::Unsubscribed);
assert_eq!(
boundary_channel_checker.state(),
ChannelState::Error("error")
);
}
#[test]
fn test_unsubscribe() {
let (mut sender, observable, channel_checker) = test_channel::<'_, _, Infallible>();
let (mut boundary_sender, boundary_observable, boundary_channel_checker) = test_channel();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = observable.window(boundary_observable);
let checker_sub_vec_cloned = checker_sub_vec.clone();
let subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(111);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(222);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(333);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
subscription.dispose();
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, sub)) in safe_lock!(mem_take: checker_sub_vec)
.into_iter()
.enumerate()
{
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
sub.dispose();
assert_eq!(checker.state(), State::Dropped);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Dropped);
assert_eq!(channel_checker.state(), ChannelState::Unsubscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Unsubscribed);
}
#[test]
fn test_ref() {
let value_1 = 111;
let value_2 = 222;
let value_3 = 333;
let error = -1;
let (mut sender, observable, channel_checker) = test_channel();
let (mut boundary_sender, boundary_observable, boundary_channel_checker) = test_channel();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = observable.window(boundary_observable);
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert!(checker.values().is_empty());
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(&value_1);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [&value_1]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [&value_1]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert!(checker.values().is_empty());
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(&value_2);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [&value_1]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [&value_2]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(&value_3);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [&value_1]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [&value_2, &value_3]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [&value_1]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [&value_2, &value_3]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert!(checker.values().is_empty());
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_termination(Termination::Error(&error));
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [&value_1]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [&value_2, &value_3]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert!(checker.values().is_empty());
assert_eq!(checker.state(), State::Error(&error));
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Error(&error));
assert_eq!(channel_checker.state(), ChannelState::Error(&error));
assert_eq!(boundary_channel_checker.state(), ChannelState::Unsubscribed);
}
#[test]
fn test_async() {
block_on(|runtime| async move {
let (mut sender, observable, channel_checker) = test_channel::<'_, _, Infallible>();
let (mut boundary_sender, boundary_observable, boundary_channel_checker) = test_channel();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = observable.window(boundary_observable);
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = runtime
.spawn(async move {
observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
)
})
.await
.unwrap();
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
let mut sender = runtime
.spawn(async move {
sender.on_next(111);
sender
})
.await
.unwrap();
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
let mut boundary_sender = runtime
.spawn(async move {
boundary_sender.on_next(());
boundary_sender
})
.await
.unwrap();
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
let mut sender = runtime
.spawn(async move {
sender.on_next(222);
sender
})
.await
.unwrap();
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
let sender = runtime
.spawn(async move {
sender.on_next(333);
sender
})
.await
.unwrap();
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
let _boundary_sender = runtime
.spawn(async move {
boundary_sender.on_next(());
boundary_sender
})
.await
.unwrap();
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
runtime
.spawn(async move { sender.on_termination(Termination::Completed) })
.await
.unwrap();
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Completed);
assert_eq!(channel_checker.state(), ChannelState::Completed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Unsubscribed);
});
}
#[test]
fn test_subscribe_by_different_observer() {
let mut subject = PublishSubject::<'_, _, Infallible>::default();
let mut boundary_subject = PublishSubject::default();
let (termination_checker_1, termination_observer_1) = Checker::<Infallible, _>::new();
let checker_sub_vec_1 = Shared::new(Mutable::new(Vec::new()));
let (termination_checker_2, termination_observer_2) = Checker::<Infallible, _>::new();
let checker_sub_vec_2 = Shared::new(Mutable::new(Vec::new()));
let observable = subject.clone();
let observable = observable.window(boundary_subject.clone());
let observable_1 = observable;
let observable_2 = observable_1.clone();
let checker_sub_vec_cloned = checker_sub_vec_1.clone();
let _subscription_1 = observable_1.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer_1.on_termination(termination);
},
);
let checker_sub_vec_cloned = checker_sub_vec_2.clone();
let _subscription_2 = observable_2.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer_2.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_1), 1);
for (index, (checker, _)) in checker_sub_vec_1.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_1.state(), State::Active);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_2), 1);
for (index, (checker, _)) in checker_sub_vec_2.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_2.state(), State::Active);
subject.on_next(111);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_1), 1);
for (index, (checker, _)) in checker_sub_vec_1.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_1.state(), State::Active);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_2), 1);
for (index, (checker, _)) in checker_sub_vec_2.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_2.state(), State::Active);
boundary_subject.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec_1), 2);
for (index, (checker, _)) in checker_sub_vec_1.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_1.state(), State::Active);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_2), 2);
for (index, (checker, _)) in checker_sub_vec_2.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_2.state(), State::Active);
subject.on_next(222);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_1), 2);
for (index, (checker, _)) in checker_sub_vec_1.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_1.state(), State::Active);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_2), 2);
for (index, (checker, _)) in checker_sub_vec_2.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_2.state(), State::Active);
subject.on_next(333);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_1), 2);
for (index, (checker, _)) in checker_sub_vec_1.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_1.state(), State::Active);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_2), 2);
for (index, (checker, _)) in checker_sub_vec_2.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_2.state(), State::Active);
boundary_subject.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec_1), 3);
for (index, (checker, _)) in checker_sub_vec_1.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_1.state(), State::Active);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_2), 3);
for (index, (checker, _)) in checker_sub_vec_2.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_2.state(), State::Active);
subject.on_termination(Termination::Completed);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_1), 3);
for (index, (checker, _)) in checker_sub_vec_1.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_1.state(), State::Completed);
assert_eq!(safe_lock_vec!(len: checker_sub_vec_2), 3);
for (index, (checker, _)) in checker_sub_vec_2.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
assert_eq!(termination_checker_2.state(), State::Completed);
}
#[test]
fn test_unsub_on_next_by_take() {
let (_sender, observable, channel_checker) = test_channel::<'_, i32, Infallible>();
let (_boundary_sender, boundary_observable, boundary_channel_checker) = test_channel();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = observable.window(boundary_observable).take(1);
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Completed);
assert_eq!(channel_checker.state(), ChannelState::Unsubscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Unsubscribed);
}
#[test]
fn test_multiple_operation() {
let (mut sender, observable, channel_checker) = test_channel::<'_, _, Infallible>();
let (mut boundary_sender_1, boundary_observable_1, boundary_channel_checker_1) = test_channel();
let (mut boundary_sender_2, boundary_observable_2, boundary_channel_checker_2) = test_channel();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let context = Shared::new(Mutable::new(Vec::new()));
let observable = observable
.window(boundary_observable_1)
.window(boundary_observable_2);
let context_cloned = context.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let checker_sub_vec_cloned = checker_sub_vec.clone();
let sub = value.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|_| {},
);
safe_lock_vec!(push: context_cloned, (checker_sub_vec, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: context), 1);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_1.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_2.state(), ChannelState::Subscribed);
sender.on_next(111);
assert_eq!(safe_lock_vec!(len: context), 1);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_1.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_2.state(), ChannelState::Subscribed);
boundary_sender_1.on_next(());
assert_eq!(safe_lock_vec!(len: context), 1);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 2);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_1.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_2.state(), ChannelState::Subscribed);
boundary_sender_2.on_next(());
assert_eq!(safe_lock_vec!(len: context), 2);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 2);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
1 => {
assert_eq!(safe_lock_vec!(len: checker), 0);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_1.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_2.state(), ChannelState::Subscribed);
sender.on_next(222);
assert_eq!(safe_lock_vec!(len: context), 2);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 2);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
1 => {
assert_eq!(safe_lock_vec!(len: checker), 0);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_1.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_2.state(), ChannelState::Subscribed);
boundary_sender_1.on_next(());
assert_eq!(safe_lock_vec!(len: context), 2);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 2);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
1 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_1.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_2.state(), ChannelState::Subscribed);
sender.on_next(333);
assert_eq!(safe_lock_vec!(len: context), 2);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 2);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
1 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_1.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker_2.state(), ChannelState::Subscribed);
sender.on_termination(Termination::Completed);
assert_eq!(safe_lock_vec!(len: context), 2);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 2);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
1 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [333]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Completed);
assert_eq!(channel_checker.state(), ChannelState::Completed);
assert_eq!(
boundary_channel_checker_1.state(),
ChannelState::Unsubscribed
);
assert_eq!(
boundary_channel_checker_2.state(),
ChannelState::Unsubscribed
);
}
#[test]
fn test_multiple_operation_same_boundary() {
let (mut sender, observable, channel_checker) = test_channel::<'_, _, Infallible>();
let mut boundary_subject = PublishSubject::default();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let context = Shared::new(Mutable::new(Vec::new()));
let observable = observable
.window(boundary_subject.clone())
.window(boundary_subject.clone());
let context_cloned = context.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let checker_sub_vec_cloned = checker_sub_vec.clone();
let sub = value.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|_| {},
);
safe_lock_vec!(push: context_cloned, (checker_sub_vec, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: context), 1);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
sender.on_next(111);
assert_eq!(safe_lock_vec!(len: context), 1);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
boundary_subject.on_next(());
assert_eq!(safe_lock_vec!(len: context), 2);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
1 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
sender.on_next(222);
assert_eq!(safe_lock_vec!(len: context), 2);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
1 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
boundary_subject.on_next(());
assert_eq!(safe_lock_vec!(len: context), 3);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
1 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
2 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
sender.on_next(333);
assert_eq!(safe_lock_vec!(len: context), 3);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
1 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
2 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
sender.on_termination(Termination::Completed);
assert_eq!(safe_lock_vec!(len: context), 3);
for (index, (checker, _)) in context.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
1 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
2 => {
assert_eq!(safe_lock_vec!(len: checker), 1);
for (index, (checker, _)) in checker.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [333]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Completed);
assert_eq!(channel_checker.state(), ChannelState::Completed);
}
#[test]
fn test_without_convenient_api() {
let (mut sender, observable, channel_checker) = test_channel::<'_, _, Infallible>();
let (mut boundary_sender, boundary_observable, boundary_channel_checker) = test_channel();
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = Window::new(observable, boundary_observable);
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(111);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(222);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_next(333);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
boundary_sender.on_next(());
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
assert_eq!(channel_checker.state(), ChannelState::Subscribed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Subscribed);
sender.on_termination(Termination::Completed);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 3);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [222, 333]);
assert_eq!(checker.state(), State::Completed);
}
2 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Completed);
assert_eq!(channel_checker.state(), ChannelState::Completed);
assert_eq!(boundary_channel_checker.state(), ChannelState::Unsubscribed);
}
#[test]
fn test_revert_completed() {
let mut subject = PublishSubject::<'_, _, Infallible>::default();
let mut boundary_subject = PublishSubject::default();
let (checker_1, observer_1) = Checker::new();
let (checker_2, observer_2) = Checker::new();
let (checker_3, observer_3) = Checker::new();
let observable = subject.clone();
let observable = observable.window(boundary_subject.clone());
let observable_1 = observable.clone().merge_all();
let observable_2 = observable.clone().concat_all();
let observable_3 = observable.switch();
let _subscription_1 = observable_1.subscribe(observer_1);
let _subscription_2 = observable_2.subscribe(observer_2);
let _subscription_3 = observable_3.subscribe(observer_3);
assert!(checker_1.values().is_empty());
assert_eq!(checker_1.state(), State::Active);
assert!(checker_2.values().is_empty());
assert_eq!(checker_2.state(), State::Active);
assert!(checker_3.values().is_empty());
assert_eq!(checker_3.state(), State::Active);
subject.on_next(111);
assert_eq!(checker_1.values(), [111]);
assert_eq!(checker_1.state(), State::Active);
assert_eq!(checker_2.values(), [111]);
assert_eq!(checker_2.state(), State::Active);
assert_eq!(checker_3.values(), [111]);
assert_eq!(checker_3.state(), State::Active);
boundary_subject.on_next(());
assert_eq!(checker_1.values(), [111]);
assert_eq!(checker_1.state(), State::Active);
assert_eq!(checker_2.values(), [111]);
assert_eq!(checker_2.state(), State::Active);
assert_eq!(checker_3.values(), [111]);
assert_eq!(checker_3.state(), State::Active);
subject.on_next(222);
assert_eq!(checker_1.values(), [111, 222]);
assert_eq!(checker_1.state(), State::Active);
assert_eq!(checker_2.values(), [111, 222]);
assert_eq!(checker_2.state(), State::Active);
assert_eq!(checker_3.values(), [111, 222]);
assert_eq!(checker_3.state(), State::Active);
subject.on_termination(Termination::Completed);
assert_eq!(checker_1.values(), [111, 222]);
assert_eq!(checker_1.state(), State::Completed);
assert_eq!(checker_2.values(), [111, 222]);
assert_eq!(checker_2.state(), State::Completed);
assert_eq!(checker_3.values(), [111, 222]);
assert_eq!(checker_3.state(), State::Completed);
}
#[test]
fn test_revert_error() {
let mut subject = PublishSubject::default();
let mut boundary_subject = PublishSubject::default();
let (checker_1, observer_1) = Checker::new();
let (checker_2, observer_2) = Checker::new();
let (checker_3, observer_3) = Checker::new();
let observable = subject.clone();
let observable = observable.window(boundary_subject.clone());
let observable_1 = observable.clone().merge_all();
let observable_2 = observable.clone().concat_all();
let observable_3 = observable.switch();
let _subscription_1 = observable_1.subscribe(observer_1);
let _subscription_2 = observable_2.subscribe(observer_2);
let _subscription_3 = observable_3.subscribe(observer_3);
assert!(checker_1.values().is_empty());
assert_eq!(checker_1.state(), State::Active);
assert!(checker_2.values().is_empty());
assert_eq!(checker_2.state(), State::Active);
assert!(checker_3.values().is_empty());
assert_eq!(checker_3.state(), State::Active);
subject.on_next(111);
assert_eq!(checker_1.values(), [111]);
assert_eq!(checker_1.state(), State::Active);
assert_eq!(checker_2.values(), [111]);
assert_eq!(checker_2.state(), State::Active);
assert_eq!(checker_3.values(), [111]);
assert_eq!(checker_3.state(), State::Active);
boundary_subject.on_next(());
assert_eq!(checker_1.values(), [111]);
assert_eq!(checker_1.state(), State::Active);
assert_eq!(checker_2.values(), [111]);
assert_eq!(checker_2.state(), State::Active);
assert_eq!(checker_3.values(), [111]);
assert_eq!(checker_3.state(), State::Active);
subject.on_next(222);
assert_eq!(checker_1.values(), [111, 222]);
assert_eq!(checker_1.state(), State::Active);
assert_eq!(checker_2.values(), [111, 222]);
assert_eq!(checker_2.state(), State::Active);
assert_eq!(checker_3.values(), [111, 222]);
assert_eq!(checker_3.state(), State::Active);
subject.on_termination(Termination::Error("error"));
assert_eq!(checker_1.values(), [111, 222]);
assert_eq!(checker_1.state(), State::Error("error"));
assert_eq!(checker_2.values(), [111, 222]);
assert_eq!(checker_2.state(), State::Error("error"));
assert_eq!(checker_3.values(), [111, 222]);
assert_eq!(checker_3.state(), State::Error("error"));
}
#[test]
fn test_immediate_next() {
let mut subject = BehaviorSubject::new(111);
let boundary_subject = BehaviorSubject::new(());
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = subject.clone().window(boundary_subject.clone());
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [111]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
subject.on_next(222);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [111, 222]);
assert_eq!(checker.state(), State::Active);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Active);
subject.on_termination(Termination::<Infallible>::Completed);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 2);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
1 => {
assert_eq!(checker.values(), [111, 222]);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Completed);
}
#[test]
fn test_immediate_completed() {
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = Empty.window(Empty.map_infallible_to_value());
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Completed);
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Completed);
}
#[test]
fn test_immediate_error() {
let (termination_checker, termination_observer) = Checker::<Infallible, _>::new();
let checker_sub_vec = Shared::new(Mutable::new(Vec::new()));
let observable = Throw::new("error").window(Throw::new("error").map_infallible_to_value());
let checker_sub_vec_cloned = checker_sub_vec.clone();
let _subscription = observable.subscribe_with_callback(
move |value| {
let (checker, observer) = Checker::new();
let sub = value.subscribe(observer);
safe_lock_vec!(push: checker_sub_vec_cloned, (checker, sub));
},
|termination| {
termination_observer.on_termination(termination);
},
);
assert_eq!(safe_lock_vec!(len: checker_sub_vec), 1);
for (index, (checker, _)) in checker_sub_vec.test_lock_ref().iter().enumerate() {
match index {
0 => {
assert_eq!(checker.values(), []);
assert_eq!(checker.state(), State::Error("error"));
}
_ => panic!(),
}
}
assert_eq!(termination_checker.state(), State::Error("error"));
}
#[test]
fn test_lifetime_sub() {
let life_marker_1 = TestStruct;
let life_marker_2 = TestStruct;
let _subscription;
{
let observable = Create::new(|mut observer| {
observer.on_next(111);
Subscription::new_with_disposal_callback(|| {
life_marker_1.consume_ref();
})
});
let boundary_subject = Create::new(|mut observer| {
observer.on_next(());
Subscription::new_with_disposal_callback(|| {
life_marker_2.consume_ref();
})
});
let observable = observable.window(boundary_subject);
let (_, observer) = Checker::<_, ()>::new();
_subscription = observable.subscribe(observer);
}
}
#[test]
fn test_lifetime_or() {
let life_marker_3 = TestStruct;
let mut life_marker_1 = None;
let mut life_marker_2 = None;
{
let observable = Create::new(|observer| {
life_marker_1 = Some(observer);
Subscription::default()
});
let boundary_subject = Create::new(|observer| {
life_marker_2 = Some(observer);
Subscription::default()
});
let observable = observable.window(boundary_subject);
let (_, mut observer) = Checker::<_, Infallible>::new();
let mut subject = PublishSubject::default();
subject.on_next(&life_marker_3);
let subject_observable = SubjectObservable::new(subject);
observer.on_next(subject_observable);
let _subscription = observable.subscribe(observer);
}
}
#[test]
fn test_lifetime_or_sub() {
let life_marker_sub = TestStruct;
let mut life_marker_or = None;
{
let observable = Create::new(|observer: BoxedObserver<'_, &TestStruct, Infallible>| {
life_marker_or = Some(observer);
Subscription::new_with_disposal_callback(|| {
life_marker_sub.consume_ref();
})
});
let boundary_subject = Create::new(|_| Subscription::default());
let observable = observable.window(boundary_subject);
let (_, observer) = Checker::new();
let _subscription = observable.subscribe(observer);
}
}
#[test]
fn test_clone() {
let observable = Create::new(|mut observer| {
observer.on_next(TestStruct);
observer.on_termination(Termination::Error(TestStruct));
Subscription::default()
});
let boundary_subject = Create::new(|_| Subscription::default());
let observable = observable.window(boundary_subject);
_ = observable.clone(); }
#[test]
fn test_type_inference_with_subscribe() {
let subject: PublishSubject<'_, i32, String> = PublishSubject::default();
let boundary_subject = PublishSubject::default();
let observable = subject.window(boundary_subject);
let observable = observable.filter(|_| true);
let (_, observer) = Checker::new();
observable.subscribe(observer);
}
#[test]
fn test_type_inference_without_subscribe() {
let subject: PublishSubject<'_, i32, String> = PublishSubject::default();
let boundary_subject: PublishSubject<'_, (), String> = PublishSubject::default();
let observable = subject.window(boundary_subject);
observable.filter(|_| true);
}