x-pipe-rs 0.1.0

Composable recommendation/feed pipeline framework built on comp-cat-rs
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

//! Candidate sources.
//!
//! A [`Source`] produces candidate items given a query.
//! Multiple sources can run in parallel via [`fan_out_2`],
//! [`fan_out_3`], and [`fan_out_4`], which use
//! [`par_zip`] from `comp-cat-rs` for concurrent execution.
//!
//! [`par_zip`]: comp_cat_rs::effect::fiber::par_zip

use comp_cat_rs::effect::fiber::{par_zip, FiberError};
use comp_cat_rs::effect::io::Io;

use crate::stage::Stage;

/// A source of candidate items.
///
/// Given a query of type `Q`, produces a `Vec<I>` of candidates
/// wrapped in [`Io`] for lazy execution.
///
/// [`Io`]: comp_cat_rs::effect::io::Io
pub trait Source<Q, I, E> {
    /// Produce candidates for a query.
    fn candidates(&self, query: &Q) -> Io<E, Vec<I>>;
}

/// Convert a [`Source`] into a [`Stage`] from `Q` to `Vec<I>`.
///
/// The source is called with a reference to the query, so the
/// query must be `Clone` to move into the stage closure.
pub fn source_stage<S, Q, I, E>(source: S) -> Stage<E, Q, Vec<I>>
where
    S: Source<Q, I, E> + Send + 'static,
    Q: Clone + Send + 'static,
    I: Send + 'static,
    E: Send + 'static,
{
    Stage::new(move |q: Q| source.candidates(&q))
}

/// Fan out to two sources in parallel, merging results.
///
/// Uses [`par_zip`] to run both sources concurrently on
/// separate OS threads.
///
/// [`par_zip`]: comp_cat_rs::effect::fiber::par_zip
#[must_use]
pub fn fan_out_2<S1, S2, Q, I, E>(
    s1: &S1,
    s2: &S2,
    query: &Q,
) -> Io<FiberError<E>, Vec<I>>
where
    S1: Source<Q, I, E>,
    S2: Source<Q, I, E>,
    E: Send + 'static,
    I: Send + 'static,
{
    let io1 = s1.candidates(query);
    let io2 = s2.candidates(query);
    par_zip(io1, io2).map(|(a, b)| {
        a.into_iter().chain(b).collect()
    })
}

/// Fan out to three sources in parallel, merging results.
#[must_use]
pub fn fan_out_3<S1, S2, S3, Q, I, E>(
    s1: &S1,
    s2: &S2,
    s3: &S3,
    query: &Q,
) -> Io<FiberError<FiberError<E>>, Vec<I>>
where
    S1: Source<Q, I, E>,
    S2: Source<Q, I, E>,
    S3: Source<Q, I, E>,
    E: Send + 'static,
    I: Send + 'static,
    FiberError<E>: Send,
{
    let io12 = fan_out_2(s1, s2, query);
    let io3 = s3.candidates(query).map_error(FiberError::Failed);
    par_zip(io12, io3).map(|(ab, c)| {
        ab.into_iter().chain(c).collect()
    })
}

/// Fan out to four sources in parallel, merging results.
#[must_use]
pub fn fan_out_4<S1, S2, S3, S4, Q, I, E>(
    s1: &S1,
    s2: &S2,
    s3: &S3,
    s4: &S4,
    query: &Q,
) -> Io<FiberError<FiberError<E>>, Vec<I>>
where
    S1: Source<Q, I, E>,
    S2: Source<Q, I, E>,
    S3: Source<Q, I, E>,
    S4: Source<Q, I, E>,
    E: Send + 'static,
    I: Send + 'static,
    FiberError<E>: Send,
{
    let io12 = fan_out_2(s1, s2, query);
    let io34 = fan_out_2(s3, s4, query);
    par_zip(io12, io34).map(|(ab, cd)| {
        ab.into_iter().chain(cd).collect()
    })
}