feature: add trigger strategy

Author: shenek

src/extract.rs

@@ -2,28 +2,34 @@ use super::{RustMatcher, StreamsonError};
 use pyo3::prelude::*;
 use streamson_lib::strategy;
 
-/// Low level Python wrapper for Simple matcher and Buffer handler
+/// Low level Python wrapper for Extract strategy
 #[pyclass]
 pub struct Extract {
     extract: strategy::Extract,
 }
 
 #[pymethods]
 impl Extract {
-    /// Create a new instance of Streamson
+    /// Create a new instance of Extract
     ///
     /// # Arguments
-    /// * `matches` - a list of valid simple matches (e.g. `{"users"}`, `[]{"name"}`, `[0]{}`)
     /// * `export_path` - indicator whether path is required in further processing
     #[new]
-    pub fn new(matcher: &RustMatcher, export_path: Option<bool>) -> PyResult<Self> {
+    pub fn new(export_path: Option<bool>) -> PyResult<Self> {
         let export_path = export_path.unwrap_or(true);
-        let mut extract = strategy::Extract::new().set_export_path(export_path);
-        extract.add_matcher(Box::new(matcher.inner.clone()));
+        let extract = strategy::Extract::new().set_export_path(export_path);
         Ok(Self { extract })
     }
 
-    /// Feeds Streamson processor with data
+    /// Adds matcher for Extract
+    ///
+    /// # Arguments
+    /// * `matcher` - matcher to be added (`Simple` or `Depth`)
+    pub fn add_matcher(&mut self, matcher: &RustMatcher) {
+        self.extract.add_matcher(Box::new(matcher.inner.clone()));
+    }
+
+    /// Process data for Extract strategy
     ///
     /// # Arguments
     /// * `data` - input data to be processed

src/lib.rs

@@ -1,5 +1,8 @@
 pub mod extract;
+pub mod trigger;
+
 pub use extract::Extract;
+pub use trigger::{PythonHandler, Trigger};
 
 use pyo3::{class::PyNumberProtocol, create_exception, exceptions, prelude::*};
 
@@ -85,6 +88,8 @@ impl PyNumberProtocol for RustMatcher {
 fn streamson(_py: Python, m: &PyModule) -> PyResult<()> {
     m.add_class::<Extract>()?;
     m.add_class::<RustMatcher>()?;
+    m.add_class::<Trigger>()?;
+    m.add_class::<PythonHandler>()?;
 
     Ok(())
 }

src/trigger.rs

@@ -0,0 +1,99 @@
+use super::{RustMatcher, StreamsonError};
+use pyo3::prelude::*;
+use std::sync::{Arc, Mutex};
+use streamson_lib::{error, handler, path::Path, strategy};
+
+/// Streamson handler which calls python callable
+#[pyclass]
+#[derive(Clone)]
+pub struct PythonHandler {
+    callable: PyObject,
+    require_path: bool,
+}
+
+#[pymethods]
+impl PythonHandler {
+    /// Create instance of PythonHandler
+    ///
+    /// # Arguments
+    /// * `callable` - python callable (3 arguments)
+    /// * `require_path` - should path be passed to handler
+    #[new]
+    pub fn new(callable: PyObject, require_path: bool) -> Self {
+        Self {
+            callable,
+            require_path,
+        }
+    }
+}
+
+impl handler::Handler for PythonHandler {
+    /// Call python function as a part of rust handler
+    ///
+    /// # Arguments
+    /// * `path` - matched path
+    /// * `matcher_idx` - index of triggered matcher
+    /// * `data` - matched data
+    fn handle(
+        &mut self,
+        path: &Path,
+        matcher_idx: usize,
+        data: Option<&[u8]>,
+    ) -> Result<(), error::Handler> {
+        let gil = Python::acquire_gil();
+        self.callable
+            .call1(
+                gil.python(),
+                (
+                    if self.require_path {
+                        Some(path.to_string())
+                    } else {
+                        None
+                    },
+                    matcher_idx,
+                    data,
+                ),
+            )
+            .map_err(|_| error::Handler::new("Calling python failed"))?;
+        Ok(())
+    }
+}
+
+/// Low level Python wrapper for Trigger strategy
+#[pyclass]
+pub struct Trigger {
+    trigger: strategy::Trigger,
+}
+
+#[pymethods]
+impl Trigger {
+    /// Create a new instance of Trigger
+    ///
+    /// # Arguments
+    /// * `export_path` - indicator whether path is required in further processing
+    #[new]
+    pub fn new() -> PyResult<Self> {
+        let trigger = strategy::Trigger::new();
+        Ok(Self { trigger })
+    }
+
+    /// Adds matcher for Trigger
+    ///
+    /// # Arguments
+    /// * `matcher` - matcher to be added (`Simple` or `Depth`)
+    pub fn add_matcher(&mut self, matcher: &RustMatcher, mut handlers: Vec<PythonHandler>) {
+        let mut res: Vec<Arc<Mutex<dyn handler::Handler>>> = vec![];
+        handlers
+            .drain(..)
+            .for_each(|e| res.push(Arc::new(Mutex::new(e))));
+        self.trigger
+            .add_matcher(Box::new(matcher.inner.clone()), &res);
+    }
+
+    pub fn process(&mut self, data: &[u8]) -> PyResult<String> {
+        match self.trigger.process(data) {
+            Err(err) => Err(StreamsonError::from(err).into()),
+            Ok(()) => Ok(String::from_utf8(data.to_vec())?),
+        }
+    }
+}

streamson/__init__.py

@@ -1,2 +1,3 @@
 from .extract import extract_async, extract_fd, extract_iter  # noqa
 from .matcher import DepthMatcher, Matcher, SimpleMatcher  # noqa
+from .trigger import trigger_async, trigger_fd, trigger_iter  # noqa

streamson/extract.py

@@ -11,15 +11,16 @@ def extract_iter(
     convert: typing.Callable[[str], typing.Any] = lambda x: x,
     require_path: bool = True,
 ) -> typing.Generator[typing.Tuple[str, typing.Any], None, None]:
-    """Extracts json from generator specified by given simple matcher
+    """Extracts json from generator specified by given matcher
     :param: input_gen: input generator
     :param: matcher: used matcher
     :param: convert: function used to convert raw data
     :param: require_path: is path required in output stream
 
     :yields: path and converted data
     """
-    extract = Extract(matcher.inner, require_path)
+    extract = Extract(require_path)
+    extract.add_matcher(matcher.inner)
     for item in input_gen:
         for path, data in extract.process(item):
             yield path, convert(data)
@@ -32,7 +33,7 @@ def extract_fd(
     convert: typing.Callable[[str], typing.Any] = lambda x: x,
     require_path: bool = True,
 ) -> typing.Generator[typing.Tuple[str, typing.Any], None, None]:
-    """Extracts json from input file specified by given simple matcher
+    """Extracts json from input file specified by given matcher
     :param: input_fd: input fd
     :param: buffer_size: how many bytes can be read from a file at once
     :param: matcher: used matcher
@@ -41,7 +42,9 @@ def extract_fd(
 
     :yields: path and converted data
     """
-    extract = Extract(matcher.inner, require_path)
+    extract = Extract(require_path)
+    extract.add_matcher(matcher.inner)
+
     input_data = input_fd.read(buffer_size)
 
     while input_data:
@@ -56,15 +59,17 @@ async def extract_async(
     convert: typing.Callable[[str], typing.Any] = lambda x: x,
     require_path: bool = True,
 ):
-    """Extracts json from given async generator specified by given simple matcher
+    """Extracts json from given async generator specified by given matcher
     :param: input_gen: input generator
     :param: matcher: used matcher
     :param: convert: function used to convert raw data
     :param: require_path: is path required in output stream
 
     :yields: path and converted data
     """
-    extract = Extract(matcher.inner, require_path)
+    extract = Extract(require_path)
+    extract.add_matcher(matcher.inner)
+
     async for input_data in input_gen:
         for path, data in extract.process(input_data):
             yield path, convert(data)

streamson/trigger.py

@@ -0,0 +1,75 @@
+import typing
+
+from streamson.streamson import PythonHandler, Trigger
+
+from .matcher import Matcher
+
+
+def trigger_iter(
+    input_gen: typing.Generator[bytes, None, None],
+    handlers: typing.List[typing.Callable[[typing.Optional[str], int, bytes], None]],
+    matcher: Matcher,
+    require_path: bool = True,
+) -> typing.Generator[bytes, None, None]:
+    """Triggers handlers on matched input
+    :param input_gen: input generator
+    :param handlers: list of handlers which should be called on match
+    :param matcher: used matcher
+    :param: require_path: is path required for handlers
+
+    :yields: input data
+    """
+    trigger = Trigger()
+    trigger.add_matcher(matcher.inner, [PythonHandler(handler, require_path) for handler in handlers])
+    for item in input_gen:
+        trigger.process(item)
+        yield item
+
+
+def trigger_fd(
+    input_fd: typing.IO[bytes],
+    handlers: typing.List[typing.Callable[[typing.Optional[str], int, bytes], None]],
+    matcher: Matcher,
+    buffer_size: int = 1024 * 1024,
+    require_path: bool = True,
+) -> typing.Generator[bytes, None, None]:
+    """Triggers handlers on matched data from a file description
+    :param input_fd: input generator
+    :param handlers: list of handlers which should be called on match
+    :param matcher: used matcher
+    :param: buffer_size: how many bytes can be read from a file at once
+    :param: require_path: is path required for handlers
+
+    :yields: input data
+    """
+    trigger = Trigger()
+    trigger.add_matcher(matcher.inner, [PythonHandler(handler, require_path) for handler in handlers])
+
+    input_data = input_fd.read(buffer_size)
+
+    while input_data:
+        trigger.process(input_data)
+        yield input_data
+        input_data = input_fd.read(buffer_size)
+
+
+async def trigger_async(
+    input_gen: typing.AsyncGenerator[bytes, None],
+    handlers: typing.List[typing.Callable[[typing.Optional[str], int, bytes], None]],
+    matcher: Matcher,
+    require_path: bool = True,
+):
+    """Triggers handlers on matched data from async generator
+    :param: input_gen: input generator
+    :param handlers: list of handlers which should be called on match
+    :param: matcher: used matcher
+    :param: require_path: is path required for handlers
+
+    :yields: input data
+    """
+    trigger = Trigger()
+    trigger.add_matcher(matcher.inner, [PythonHandler(handler, require_path) for handler in handlers])
+
+    async for input_data in input_gen:
+        trigger.process(input_data)
+        yield input_data

tests/conftest.py

@@ -0,0 +1,44 @@
+import io
+import json
+import typing
+
+import pytest
+
+DATA_JSON = {
+    "users": ["john", "carl", "bob"],
+    "groups": ["admins", "users"],
+}
+
+
+@pytest.fixture(scope="function")
+def make_async_gen():
+    def make():
+        async def async_in():
+
+            yield b'{"users": '
+            yield b'["john", "carl", "bob"'
+            yield b"]}"
+
+        return async_in
+
+    return make
+
+
+@pytest.fixture
+def data() -> typing.List[bytes]:
+    return [json.dumps(DATA_JSON).encode()]
+
+
+@pytest.fixture
+def io_reader() -> io.BytesIO:
+    return io.BytesIO(json.dumps(DATA_JSON).encode())
+
+
+@pytest.fixture(scope="function")
+def handler():
+    output: typing.List[typing.Tuple[typing.Optional[str], int, bytes]] = []
+
+    def store_to_output(path: typing.Optional[str], matcher_idx: int, data: bytes):
+        output.append((path, matcher_idx, data))
+
+    return store_to_output, output

tests/test_extract.py

@@ -1,34 +1,17 @@
-import io
 import json
-import typing
 from enum import Enum, auto
 
 import hyperjson
 import pytest
 
 import streamson
 
-DATA_JSON = {
-    "users": ["john", "carl", "bob"],
-    "groups": ["admins", "users"],
-}
-
 
 class Kind(Enum):
     FD = auto()
     ITER = auto()
 
 
-@pytest.fixture
-def data() -> typing.List[bytes]:
-    return [json.dumps(DATA_JSON).encode()]
-
-
-@pytest.fixture
-def io_reader() -> io.BytesIO:
-    return io.BytesIO(json.dumps(DATA_JSON).encode())
-
-
 @pytest.mark.parametrize(
     "kind,convert,extract_path",
     [