The output half of the I/O pipeline.

Implemented the output half of the I/O framework.  Similar to input pipelines,
an output pipeline consists of an (output) transport endpoint and an encoder
that serializes output batches into a particular format.

Author: ryzhyk

Cargo.lock

@@ -21,3 +21,5 @@ bincode = { version = "2.0.0-rc.2", features = ["serde"] }
 serde_json = "1.0.87"
 size-of = { version = "0.1.2", features = ["time-std"]}
 tempfile = "3.3.0"
+proptest = "1.0.0"
+proptest-derive = "0.3.0"

adapters/Cargo.toml

@@ -1,4 +1,4 @@
-use crate::{DeCollectionHandle, DeZSetHandle};
+use crate::{DeCollectionHandle, DeZSetHandle, SerOutputBatchHandle};
 use dbsp::{algebra::ZRingValue, CollectionHandle, DBData, DBWeight};
 use serde::Deserialize;
 use std::collections::BTreeMap;
@@ -13,7 +13,8 @@ use std::collections::BTreeMap;
 /// method).
 #[derive(Default)]
 pub struct Catalog {
-    input_collections: BTreeMap<String, Box<dyn DeCollectionHandle>>,
+    input_collection_handles: BTreeMap<String, Box<dyn DeCollectionHandle>>,
+    output_batch_handles: BTreeMap<String, Box<dyn SerOutputBatchHandle>>,
 }
 
 impl Catalog {
@@ -22,25 +23,39 @@ impl Catalog {
         Self::default()
     }
 
-    pub fn register_input_zset<K, R>(&mut self, name: &str, handle: CollectionHandle<K, R>)
+    pub fn register_input_zset_handle<K, R>(&mut self, name: &str, handle: CollectionHandle<K, R>)
     where
         K: DBData + for<'de> Deserialize<'de>,
         R: DBWeight + ZRingValue,
     {
-        self.register_input(name, DeZSetHandle::new(handle));
+        self.register_input_collection_handle(name, DeZSetHandle::new(handle));
     }
 
     /// Add a named input stream handle to the catalog.
-    pub fn register_input<H>(&mut self, name: &str, handle: H)
+    pub fn register_input_collection_handle<H>(&mut self, name: &str, handle: H)
     where
         H: DeCollectionHandle + 'static,
     {
-        self.input_collections
+        self.input_collection_handles
+            .insert(name.to_owned(), Box::new(handle));
+    }
+
+    /// Add a named output stream handle to the catalog.
+    pub fn register_output_batch_handle<H>(&mut self, name: &str, handle: H)
+    where
+        H: SerOutputBatchHandle + 'static,
+    {
+        self.output_batch_handles
             .insert(name.to_owned(), Box::new(handle));
     }
 
     /// Look up an input stream handle by name.
-    pub fn input_collection(&self, name: &str) -> Option<&dyn DeCollectionHandle> {
-        self.input_collections.get(name).map(|b| &**b)
+    pub fn input_collection_handle(&self, name: &str) -> Option<&dyn DeCollectionHandle> {
+        self.input_collection_handles.get(name).map(|b| &**b)
+    }
+
+    /// Look up an output stream handle by name.
+    pub fn output_batch_handle(&self, name: &str) -> Option<&dyn SerOutputBatchHandle> {
+        self.output_batch_handles.get(name).map(|b| &**b)
     }
 }

adapters/src/catalog.rs

@@ -27,6 +27,10 @@ pub struct ControllerConfig {
 
     /// Input endpoint configuration.
     pub inputs: BTreeMap<Cow<'static, str>, InputEndpointConfig>,
+
+    /// Output endpoint configuration.
+    #[serde(default)]
+    pub outputs: BTreeMap<Cow<'static, str>, OutputEndpointConfig>,
 }
 
 /// Internal representaion of controller configuration stored inside the
@@ -36,13 +40,15 @@ pub struct ControllerConfig {
 pub struct ControllerInnerConfig {
     pub global: GlobalControllerConfig,
     pub inputs: BTreeMap<EndpointId, InputEndpointConfig>,
+    pub outputs: BTreeMap<EndpointId, OutputEndpointConfig>,
 }
 
 impl ControllerInnerConfig {
     pub fn new(global: GlobalControllerConfig) -> Self {
         Self {
             global,
             inputs: BTreeMap::new(),
+            outputs: BTreeMap::new(),
         }
     }
 }
@@ -87,24 +93,47 @@ pub struct InputEndpointConfig {
     pub max_buffered_records: u64,
 }
 
+#[derive(Clone, Serialize, Deserialize)]
+pub struct OutputEndpointConfig {
+    /// The name of the output stream of the circuit that this pipeline is
+    /// connected to.
+    pub stream: Cow<'static, str>,
+
+    /// Transport endpoint configuration.
+    pub transport: TransportConfig,
+
+    /// Encoder configuration.
+    pub format: FormatConfig,
+
+    /// Backpressure threshold.
+    ///
+    /// The default is 1 million.
+    #[serde(default = "default_max_buffered_records")]
+    pub max_buffered_records: u64,
+}
+
 /// Transport endpoint configuration.
 #[derive(Clone, Serialize, Deserialize)]
 pub struct TransportConfig {
     /// Data transport name, e.g., "file", "kafka", "kinesis", etc.
     pub name: Cow<'static, str>,
 
     /// Transport-specific endpoint configuration passed to
-    /// [`Transport::new_endpoint`](`crate::Transport::new_endpoint`).
+    /// [`OutputTransport::new_endpoint`](`crate::OutputTransport::new_endpoint`)
+    /// and
+    /// [`InputTransport::new_endpoint`](`crate::InputTransport::new_endpoint`).
+    #[serde(default)]
     pub config: YamlValue,
 }
 
 /// Data format specification used to parse raw data received from the
-/// endpoint.
+/// endpoint or to encode data sent to the endpoint.
 #[derive(Clone, Serialize, Deserialize)]
 pub struct FormatConfig {
     /// Format name, e.g., "csv", "json", "bincode", etc.
     pub name: Cow<'static, str>,
 
-    /// Format-specific parser configuration.
+    /// Format-specific parser or encoder configuration.
+    #[serde(default)]
     pub config: YamlValue,
 }