Mini cleanup

Signed-off-by: Heinz N. Gies <[email protected]>

tremor-connectors/src/source.rs

@@ -1254,9 +1254,9 @@ async fn build_events(
                 )
                 .await;
                 let (port, payload) = match line_value {
-                    Ok(decoded) => (port.unwrap_or(&OUT).clone(), decoded),
-                    Err(None) => continue,
-                    Err(Some(e)) => (
+                    Ok(Some(decoded)) => (port.unwrap_or(&OUT).clone(), decoded),
+                    Ok(None) => continue,
+                    Err(e) => (
                         ERR,
                         make_error(alias, &e.into(), stream_state.stream_id, pull_id, meta),
                     ),
@@ -1315,9 +1315,9 @@ async fn build_last_events(
                 )
                 .await;
                 let (port, payload) = match line_value {
-                    Ok(decoded) => (port.unwrap_or(&OUT).clone(), decoded),
-                    Err(None) => continue,
-                    Err(Some(e)) => (
+                    Ok(Some(decoded)) => (port.unwrap_or(&OUT).clone(), decoded),
+                    Ok(None) => continue,
+                    Err(e) => (
                         ERR,
                         make_error(alias, &e.into(), stream_state.stream_id, pull_id, meta),
                     ),

tremor-script/src/srs.rs

@@ -139,22 +139,18 @@ impl EventPayload {
         meta: Value<'static>,
         ingest_ns: &mut u64,
         codec: &mut Box<dyn Codec>,
-    ) -> std::result::Result<Self, Option<crate::errors::Error>> {
+    ) -> std::result::Result<Option<Self>, crate::errors::Error> {
         let mut raw = Pin::new(raw);
         // We are keeping the Vector pinnd and as part of the same struct, and the decoded data
         // can't leak this function aside of inside this struct.
         // ALLOW: See above explenation
         let r = unsafe { mem::transmute::<&mut [u8], &'static mut [u8]>(raw.as_mut().get_mut()) };
-        let res = codec.decode(r, *ingest_ns, meta).await;
-        match res {
-            Ok(None) => Err(None),
-            Err(e) => Err(Some(e.into())),
-            Ok(Some((decoded, meta))) => {
-                let data = ValueAndMeta::from_parts(decoded, meta);
-                let raw = vec![Arc::new(raw)];
-                Ok(Self { raw, data })
-            }
-        }
+        let res = codec.decode(r, *ingest_ns, meta).await?;
+        Ok(res.map(|(decoded, meta)| {
+            let data = ValueAndMeta::from_parts(decoded, meta);
+            let raw = vec![Arc::new(raw)];
+            Self { raw, data }
+        }))
     }
 
     /// Named after the original rental struct for easy rewriting.