mod equivalence {
#![allow(deprecated)]
use datum::{Flow, Source, StreamError, StreamResult};
fn boom(label: &str) -> StreamError {
StreamError::Failed(label.to_owned())
}
fn assert_equivalent(
canonical: StreamResult<Vec<i32>>,
deprecated: StreamResult<Vec<i32>>,
expected: StreamResult<Vec<i32>>,
) {
assert_eq!(&canonical, &expected);
assert_eq!(&deprecated, &expected);
assert_eq!(canonical, deprecated);
}
#[test]
fn flow_try_aliases_match_result_aliases_on_happy_and_error_paths() {
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_map(|item| Ok(item * 2)))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().map_result(|item| Ok(item * 2)))
.run_collect(),
Ok(vec![2, 4, 6]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_map(|item| {
if item == 2 {
Err(boom("flow try_map"))
} else {
Ok(item)
}
}))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().map_result(|item| {
if item == 2 {
Err(boom("flow try_map"))
} else {
Ok(item)
}
}))
.run_collect(),
Err(boom("flow try_map")),
);
assert_equivalent(
Source::from_iter([1, 2, 3, 4])
.via(Flow::identity().try_filter(|item| Ok(item % 2 == 0)))
.run_collect(),
Source::from_iter([1, 2, 3, 4])
.via(Flow::identity().filter_result(|item| Ok(item % 2 == 0)))
.run_collect(),
Ok(vec![2, 4]),
);
assert_equivalent(
Source::from_iter([1, 2, 3, 4])
.via(Flow::identity().try_filter(|item| {
if *item == 3 {
Err(boom("flow try_filter"))
} else {
Ok(item % 2 == 0)
}
}))
.run_collect(),
Source::from_iter([1, 2, 3, 4])
.via(Flow::identity().filter_result(|item| {
if *item == 3 {
Err(boom("flow try_filter"))
} else {
Ok(item % 2 == 0)
}
}))
.run_collect(),
Err(boom("flow try_filter")),
);
assert_equivalent(
Source::from_iter([1, 2, 3, 4])
.via(
Flow::identity()
.try_filter_map(|item| Ok((item % 2 == 1).then_some(item * 10))),
)
.run_collect(),
Source::from_iter([1, 2, 3, 4])
.via(
Flow::identity()
.filter_map_result(|item| Ok((item % 2 == 1).then_some(item * 10))),
)
.run_collect(),
Ok(vec![10, 30]),
);
assert_equivalent(
Source::from_iter([1, 2, 3, 4])
.via(Flow::identity().try_filter_map(|item| {
if item == 4 {
Err(boom("flow try_filter_map"))
} else {
Ok((item % 2 == 1).then_some(item * 10))
}
}))
.run_collect(),
Source::from_iter([1, 2, 3, 4])
.via(Flow::identity().filter_map_result(|item| {
if item == 4 {
Err(boom("flow try_filter_map"))
} else {
Ok((item % 2 == 1).then_some(item * 10))
}
}))
.run_collect(),
Err(boom("flow try_filter_map")),
);
assert_equivalent(
Source::from_iter([1, 3])
.via(Flow::identity().try_map_concat(|item| Ok([item, item * 10])))
.run_collect(),
Source::from_iter([1, 3])
.via(Flow::identity().map_concat_result(|item| Ok([item, item * 10])))
.run_collect(),
Ok(vec![1, 10, 3, 30]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_map_concat(|item| {
if item == 3 {
Err(boom("flow try_map_concat"))
} else {
Ok([item, item * 10])
}
}))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().map_concat_result(|item| {
if item == 3 {
Err(boom("flow try_map_concat"))
} else {
Ok([item, item * 10])
}
}))
.run_collect(),
Err(boom("flow try_map_concat")),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_fold(0, |acc, item| Ok(acc + item)))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().fold_result(0, |acc, item| Ok(acc + item)))
.run_collect(),
Ok(vec![6]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_fold(0, |acc, item| {
if item == 3 {
Err(boom("flow try_fold"))
} else {
Ok(acc + item)
}
}))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().fold_result(0, |acc, item| {
if item == 3 {
Err(boom("flow try_fold"))
} else {
Ok(acc + item)
}
}))
.run_collect(),
Err(boom("flow try_fold")),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_reduce(|acc, item| Ok(acc + item)))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().reduce_result(|acc, item| Ok(acc + item)))
.run_collect(),
Ok(vec![6]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_reduce(|acc, item| {
if item == 3 {
Err(boom("flow try_reduce"))
} else {
Ok(acc + item)
}
}))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().reduce_result(|acc, item| {
if item == 3 {
Err(boom("flow try_reduce"))
} else {
Ok(acc + item)
}
}))
.run_collect(),
Err(boom("flow try_reduce")),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_scan(0, |acc, item| Ok(acc + item)))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().scan_result(0, |acc, item| Ok(acc + item)))
.run_collect(),
Ok(vec![0, 1, 3, 6]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_scan(0, |acc, item| {
if item == 3 {
Err(boom("flow try_scan"))
} else {
Ok(acc + item)
}
}))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().scan_result(0, |acc, item| {
if item == 3 {
Err(boom("flow try_scan"))
} else {
Ok(acc + item)
}
}))
.run_collect(),
Err(boom("flow try_scan")),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_stateful_map(0, |state, item| {
*state += item;
Ok(*state)
}))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().stateful_map_result(0, |state, item| {
*state += item;
Ok(*state)
}))
.run_collect(),
Ok(vec![1, 3, 6]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_stateful_map(0, |state, item| {
if item == 3 {
Err(boom("flow try_stateful_map"))
} else {
*state += item;
Ok(*state)
}
}))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(Flow::identity().stateful_map_result(0, |state, item| {
if item == 3 {
Err(boom("flow try_stateful_map"))
} else {
*state += item;
Ok(*state)
}
}))
.run_collect(),
Err(boom("flow try_stateful_map")),
);
assert_equivalent(
Source::from_iter([1, 2])
.via(Flow::identity().try_stateful_map_concat(0, |state, item| {
*state += item;
Ok([*state, *state * 10])
}))
.run_collect(),
Source::from_iter([1, 2])
.via(
Flow::identity().stateful_map_concat_result(0, |state, item| {
*state += item;
Ok([*state, *state * 10])
}),
)
.run_collect(),
Ok(vec![1, 10, 3, 30]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.via(Flow::identity().try_stateful_map_concat(0, |state, item| {
if item == 3 {
Err(boom("flow try_stateful_map_concat"))
} else {
*state += item;
Ok([*state, *state * 10])
}
}))
.run_collect(),
Source::from_iter([1, 2, 3])
.via(
Flow::identity().stateful_map_concat_result(0, |state, item| {
if item == 3 {
Err(boom("flow try_stateful_map_concat"))
} else {
*state += item;
Ok([*state, *state * 10])
}
}),
)
.run_collect(),
Err(boom("flow try_stateful_map_concat")),
);
}
#[test]
fn source_try_aliases_match_result_aliases_on_happy_and_error_paths() {
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_map(|item| Ok(item * 2))
.run_collect(),
Source::from_iter([1, 2, 3])
.map_result(|item| Ok(item * 2))
.run_collect(),
Ok(vec![2, 4, 6]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_map(|item| {
if item == 2 {
Err(boom("source try_map"))
} else {
Ok(item)
}
})
.run_collect(),
Source::from_iter([1, 2, 3])
.map_result(|item| {
if item == 2 {
Err(boom("source try_map"))
} else {
Ok(item)
}
})
.run_collect(),
Err(boom("source try_map")),
);
assert_equivalent(
Source::from_iter([1, 2, 3, 4])
.try_filter(|item| Ok(item % 2 == 0))
.run_collect(),
Source::from_iter([1, 2, 3, 4])
.filter_result(|item| Ok(item % 2 == 0))
.run_collect(),
Ok(vec![2, 4]),
);
assert_equivalent(
Source::from_iter([1, 2, 3, 4])
.try_filter(|item| {
if *item == 3 {
Err(boom("source try_filter"))
} else {
Ok(item % 2 == 0)
}
})
.run_collect(),
Source::from_iter([1, 2, 3, 4])
.filter_result(|item| {
if *item == 3 {
Err(boom("source try_filter"))
} else {
Ok(item % 2 == 0)
}
})
.run_collect(),
Err(boom("source try_filter")),
);
assert_equivalent(
Source::from_iter([1, 2, 3, 4])
.try_filter_map(|item| Ok((item % 2 == 1).then_some(item * 10)))
.run_collect(),
Source::from_iter([1, 2, 3, 4])
.filter_map_result(|item| Ok((item % 2 == 1).then_some(item * 10)))
.run_collect(),
Ok(vec![10, 30]),
);
assert_equivalent(
Source::from_iter([1, 2, 3, 4])
.try_filter_map(|item| {
if item == 4 {
Err(boom("source try_filter_map"))
} else {
Ok((item % 2 == 1).then_some(item * 10))
}
})
.run_collect(),
Source::from_iter([1, 2, 3, 4])
.filter_map_result(|item| {
if item == 4 {
Err(boom("source try_filter_map"))
} else {
Ok((item % 2 == 1).then_some(item * 10))
}
})
.run_collect(),
Err(boom("source try_filter_map")),
);
assert_equivalent(
Source::from_iter([1, 3])
.try_map_concat(|item| Ok([item, item * 10]))
.run_collect(),
Source::from_iter([1, 3])
.map_concat_result(|item| Ok([item, item * 10]))
.run_collect(),
Ok(vec![1, 10, 3, 30]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_map_concat(|item| {
if item == 3 {
Err(boom("source try_map_concat"))
} else {
Ok([item, item * 10])
}
})
.run_collect(),
Source::from_iter([1, 2, 3])
.map_concat_result(|item| {
if item == 3 {
Err(boom("source try_map_concat"))
} else {
Ok([item, item * 10])
}
})
.run_collect(),
Err(boom("source try_map_concat")),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_fold(0, |acc, item| Ok(acc + item))
.run_collect(),
Source::from_iter([1, 2, 3])
.fold_result(0, |acc, item| Ok(acc + item))
.run_collect(),
Ok(vec![6]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_fold(0, |acc, item| {
if item == 3 {
Err(boom("source try_fold"))
} else {
Ok(acc + item)
}
})
.run_collect(),
Source::from_iter([1, 2, 3])
.fold_result(0, |acc, item| {
if item == 3 {
Err(boom("source try_fold"))
} else {
Ok(acc + item)
}
})
.run_collect(),
Err(boom("source try_fold")),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_reduce(|acc, item| Ok(acc + item))
.run_collect(),
Source::from_iter([1, 2, 3])
.reduce_result(|acc, item| Ok(acc + item))
.run_collect(),
Ok(vec![6]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_reduce(|acc, item| {
if item == 3 {
Err(boom("source try_reduce"))
} else {
Ok(acc + item)
}
})
.run_collect(),
Source::from_iter([1, 2, 3])
.reduce_result(|acc, item| {
if item == 3 {
Err(boom("source try_reduce"))
} else {
Ok(acc + item)
}
})
.run_collect(),
Err(boom("source try_reduce")),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_scan(0, |acc, item| Ok(acc + item))
.run_collect(),
Source::from_iter([1, 2, 3])
.scan_result(0, |acc, item| Ok(acc + item))
.run_collect(),
Ok(vec![0, 1, 3, 6]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_scan(0, |acc, item| {
if item == 3 {
Err(boom("source try_scan"))
} else {
Ok(acc + item)
}
})
.run_collect(),
Source::from_iter([1, 2, 3])
.scan_result(0, |acc, item| {
if item == 3 {
Err(boom("source try_scan"))
} else {
Ok(acc + item)
}
})
.run_collect(),
Err(boom("source try_scan")),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_stateful_map(0, |state, item| {
*state += item;
Ok(*state)
})
.run_collect(),
Source::from_iter([1, 2, 3])
.stateful_map_result(0, |state, item| {
*state += item;
Ok(*state)
})
.run_collect(),
Ok(vec![1, 3, 6]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_stateful_map(0, |state, item| {
if item == 3 {
Err(boom("source try_stateful_map"))
} else {
*state += item;
Ok(*state)
}
})
.run_collect(),
Source::from_iter([1, 2, 3])
.stateful_map_result(0, |state, item| {
if item == 3 {
Err(boom("source try_stateful_map"))
} else {
*state += item;
Ok(*state)
}
})
.run_collect(),
Err(boom("source try_stateful_map")),
);
assert_equivalent(
Source::from_iter([1, 2])
.try_stateful_map_concat(0, |state, item| {
*state += item;
Ok([*state, *state * 10])
})
.run_collect(),
Source::from_iter([1, 2])
.stateful_map_concat_result(0, |state, item| {
*state += item;
Ok([*state, *state * 10])
})
.run_collect(),
Ok(vec![1, 10, 3, 30]),
);
assert_equivalent(
Source::from_iter([1, 2, 3])
.try_stateful_map_concat(0, |state, item| {
if item == 3 {
Err(boom("source try_stateful_map_concat"))
} else {
*state += item;
Ok([*state, *state * 10])
}
})
.run_collect(),
Source::from_iter([1, 2, 3])
.stateful_map_concat_result(0, |state, item| {
if item == 3 {
Err(boom("source try_stateful_map_concat"))
} else {
*state += item;
Ok([*state, *state * 10])
}
})
.run_collect(),
Err(boom("source try_stateful_map_concat")),
);
}
}