kproc 0.7.0

Knowledge Processing library.
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
112
113
114
115
116

//! Traits for defining knowledge processing units.

pub use kproc_pmacros::{InputStreams, OutputStreams};

use std::future::Future;

use crate::{utils, Result};

/// Input values for streams
pub trait Inputs
{
  /// Type of the input streams
  type Streams: InputStreams<Inputs = Self>;
}

/// Output values for streams
pub trait Outputs
{
  /// Type of the output streams
  type Streams: OutputStreams<Outputs = Self>;
}
/// Input streams for processors
pub trait InputStreams
{
  /// Type of the input value
  type Inputs: Inputs<Streams = Self>;
  /// Get the next input value
  fn next(&mut self) -> impl Future<Output = Result<Self::Inputs>>;
}

/// Output streams for processors
pub trait OutputStreams
{
  /// Type of the output value
  type Outputs: Outputs<Streams = Self>;
  /// Check if one of the output channel is full.
  fn is_full(&self) -> bool;
  /// Send the next value
  fn next(&mut self, outputs: Self::Outputs) -> impl Future<Output = crate::Result<()>>;
}

/// A processor is a computation unit in a knowledge processing stream.
pub trait ProcessorStreams
{
  /// A tuple of input values
  type Inputs;
  /// A tuple of output values
  type Outputs;
  /// A tuple of input streams
  type InputStreams: InputStreams<Inputs = Self::Inputs>;
  /// A tuple of output streams
  type OutputStreams: OutputStreams<Outputs = Self::Outputs>;
}

/// Trait for implementing processors that process value one-by-one
pub trait ValueProcessor: ProcessorStreams
{
  /// Process an input
  fn process_one(&self, inputs: Self::Inputs)
    -> impl Future<Output = Result<Self::Outputs>> + Send;
}

/// Process input data, from streams into output streams
pub trait Processor: ProcessorStreams
where
  Self: Sized,
{
  /// start processing input streams into output streams
  fn start(
    self,
    inputs: Self::InputStreams,
    outputs: Self::OutputStreams,
  ) -> impl Future<Output = ()>;
}

impl<T> Processor for T
where
  T: ValueProcessor,
{
  #[allow(clippy::manual_async_fn)] // https://github.com/rust-lang/rust-clippy/issues/12664
  fn start(
    self,
    mut inputs: Self::InputStreams,
    mut outputs: Self::OutputStreams,
  ) -> impl Future<Output = ()>
  {
    async move {
      loop
      {
        let r = inputs.next().await;
        match r
        {
          Ok(i) =>
          {
            let output_values = self.process_one(i).await;
            match output_values
            {
              Ok(ov) =>
              {
                utils::log_error(outputs.next(ov).await);
              }
              Err(_) =>
              {
                utils::log_error(output_values);
              }
            }
          }
          Err(_) =>
          {
            return;
          }
        }
      }
    }
  }
}