WIP - Feature/smart module filter

Cargo.toml

@@ -15,5 +15,6 @@ log = "^0.4.17"
 
 [dependencies]
 cpython = { version = "0.7", features = ["extension-module"] }
+flate2 = "1.0.24"
 fluvio = { version = "0.13" }
-fluvio-future = { version = "0.4.2", features = ["task", "io"] }
+fluvio-future = { version = "0.4.2", features = ["task", "io"] }

fluvio/__init__.py

@@ -140,6 +140,35 @@ def stream(self, offset: Offset) -> PartitionConsumerStream:
         '''
         return PartitionConsumerStream(self._inner.stream(offset._inner))
 
+    def stream_with_config(self, offset: Offset, wasm_module_path: str) -> PartitionConsumerStream:
+        '''
+        Continuously streams events from a particular offset with a SmartModule WASM module in the consumer’s
+        partition. This returns a `PartitionConsumerStream` which is an
+        iterator.
+
+        Streaming is one of the two ways to consume events in Fluvio. It is a
+        continuous request for new records arriving in a partition, beginning
+        at a particular offset. You specify the starting point of the stream
+        using an Offset and periodically receive events, either individually or
+        in batches.
+
+        Args:
+            offset: Offset
+            wasm_module_path: str - The absolute path to the WASM file
+
+        Example:
+            import os
+
+            wasm_module_path = os.path.abspath("somefilter.wasm")
+            for i in consumer.stream_with_config(Offset.beginning(), wasm_module_path):
+                # do something with i
+
+        Returns:
+            PartionConsumerStream
+
+        '''
+        return PartitionConsumerStream(self._inner.stream_with_config(offset._inner, wasm_module_path))
+
 
 class TopicProducer:
     '''An interface for producing events to a particular topic.

integration-tests/test_fluvio_python.py

@@ -1,6 +1,8 @@
+from string import ascii_lowercase
 from fluvio import (Fluvio, FluviorError, Offset)
 import unittest
 import uuid
+import os
 
 
 def create_topic(topic):
@@ -32,7 +34,34 @@ def test_produce(self):
         for i in range(10):
             producer.send_string("FOOBAR %s " % i)
 
-    def test_consume_with_iterator(self):
+    def test_consume_with_smart_module_iterator(self):
+        """
+        Test adds a the alphabet into a topic in the format of record-[letter]
+
+        A wasm smart module is added to the filter and a all messages are retrieved and stored in the records list
+        We can then assert the following:
+
+        - There should be 1 item
+        - It should be record-a
+
+        """
+
+        wasm_module_path = os.path.abspath("integration-tests/smartmodule_filter_on_a.wasm")
+
+        fluvio = Fluvio.connect()
+        producer = fluvio.topic_producer(self.topic)
+        for i in list(ascii_lowercase):
+            producer.send_string(f"record-{i}")
+
+        records = []
+
+        consumer = fluvio.partition_consumer(self.topic, 0)
+        records.append(bytearray(next(consumer.stream_with_config(Offset.beginning(), wasm_module_path)).value()).decode())
+
+        self.assertEqual(len(records), 1)
+        self.assertEqual(records[0], "record-a")
+
+    def test_consumer_with_interator(self):
         fluvio = Fluvio.connect()
         producer = fluvio.topic_producer(self.topic)
         for i in range(10):

src/glue.rs.in

@@ -7,13 +7,22 @@ use fluvio::{
     Offset,
     consumer::Record,
 };
+use std::io::Read;
+use flate2::bufread::GzEncoder;
+use flate2::Compression;
 use fluvio_future::{
     task::run_block_on,
     io::{
         Stream,
         StreamExt,
     },
 };
+use fluvio::consumer::{
+    SmartModuleInvocation,
+    SmartModuleInvocationWasm,
+    SmartModuleKind,
+    ConsumerConfig
+};
 use std::pin::Pin;
 
 mod _Fluvio {
@@ -52,6 +61,22 @@ foreign_class!(class Fluvio {
 });
 
 
+pub struct ConsumerConfigWrapper {
+    wasm_module: Vec<u8>,
+}
+
+impl ConsumerConfigWrapper {
+    fn new_config_with_wasm_filter(file: &str) -> ConsumerConfigWrapper {
+        let raw_buffer = std::fs::read(file);
+        let mut encoder = GzEncoder::new(raw_buffer.as_ref().unwrap().as_slice(), Compression::default());
+        let mut buffer = Vec::with_capacity(raw_buffer.as_ref().unwrap().len());
+        encoder.read_to_end(&mut buffer);
+        ConsumerConfigWrapper {
+            wasm_module: buffer,
+        }
+    }
+}
+
 mod _PartitionConsumer {
     use super::*;
     pub fn stream(
@@ -62,13 +87,31 @@ mod _PartitionConsumer {
             inner: Box::pin(run_block_on(consumer.stream(offset.clone()))?)
         })
     }
+    pub fn stream_with_config(
+        consumer: &PartitionConsumer,
+        offset: &Offset,
+        wasm_module_path: &str
+    ) -> Result<PartitionConsumerStream, FluvioError> {
+        let config_wrapper = ConsumerConfigWrapper::new_config_with_wasm_filter(wasm_module_path);
+        let mut builder = ConsumerConfig::builder();
+        builder.smartmodule(Some(SmartModuleInvocation {
+            wasm: SmartModuleInvocationWasm::AdHoc(config_wrapper.wasm_module),
+            kind: SmartModuleKind::Filter,
+            params: Default::default()
+        }));
+        let config = builder.build().map_err(|err| FluvioError::Other(err.to_string()))?;
+        run_block_on(consumer.stream_with_config(offset.clone(), config)).map(|stream| PartitionConsumerStream { inner: Box::pin(stream) })
+    }
 }
 foreign_class!(class PartitionConsumer {
     self_type PartitionConsumer;
     private constructor = empty;
     fn _PartitionConsumer::stream(
         &self, _: &Offset
     ) -> Result<PartitionConsumerStream, FluvioError>;
+    fn _PartitionConsumer::stream_with_config(
+        &self, _: &Offset, _: &str 
+    ) -> Result<PartitionConsumerStream, FluvioError>;
 });
 
 type PartitionConsumerIteratorInner =

src/lib.rs

@@ -1,3 +1,3 @@
 #![allow(non_snake_case, unused)]
 
-include!(concat!(env!("OUT_DIR"), "/glue.rs"));
+include!(concat!(env!("OUT_DIR"), "/glue.rs"));