WIP

crates/streamio/src/lib.rs

@@ -0,0 +1,3 @@
+pub mod streamtimeout;
+pub mod tcprawclient;
+pub mod tcpreadasbytes;

crates/streamio/src/streamtimeout.rs

@@ -0,0 +1,23 @@
+use futures_util::Stream;
+use std::pin::Pin;
+use streams::streamtimeout::TimeoutableStream;
+
+pub struct StreamTimeout {}
+
+impl StreamTimeout {
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl<S> streams::streamtimeout::StreamTimeout<S> for StreamTimeout {
+    fn timeout_intervals(&self, inp: Pin<Box<dyn Stream<Item = S> + Send>>) -> Pin<Box<dyn Stream<Item = S> + Send>> {
+        todo!()
+    }
+}
+
+impl<S> streams::streamtimeout::StreamTimeout2<S> for StreamTimeout {
+    fn timeout_intervals(&self, inp: S) -> TimeoutableStream<S> {
+        todo!()
+    }
+}

crates/streamio/src/tcprawclient.rs

@@ -0,0 +1,111 @@
+use crate::tcpreadasbytes::TcpReadAsBytes;
+use futures_util::Stream;
+use futures_util::TryStreamExt;
+use items_0::framable::FrameTypeInnerStatic;
+use items_0::streamitem::sitem_data;
+use items_0::streamitem::sitem_err2_from_string;
+use items_0::streamitem::Sitemty;
+use items_2::eventfull::EventFull;
+use items_2::framable::Framable;
+use items_2::frame::make_term_frame;
+use netpod::log::*;
+use netpod::Cluster;
+use netpod::Node;
+use netpod::ReqCtx;
+use query::api4::events::EventsSubQuery;
+use serde::de::DeserializeOwned;
+use std::fmt;
+use std::pin::Pin;
+use streams::frames::eventsfromframes::EventsFromFrames;
+use streams::frames::inmem::BoxedBytesStream;
+use streams::frames::inmem::InMemoryFrameStream;
+use streams::tcprawclient::make_node_command_frame;
+use streams::tcprawclient::x_processed_event_blobs_stream_from_node_http;
+use streams::tcprawclient::HttpSimplePost;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpStream;
+
+#[derive(Debug, thiserror::Error)]
+#[cstm(name = "TcpRawClient")]
+pub enum Error {
+    IO(#[from] std::io::Error),
+    Frame(#[from] items_2::frame::Error),
+    Framable(#[from] items_2::framable::Error),
+    StreamsRawClient(#[from] streams::tcprawclient::Error),
+}
+
+pub type BoxedStream<T> = Pin<Box<dyn Stream<Item = Sitemty<T>> + Send>>;
+
+pub async fn x_processed_event_blobs_stream_from_node_tcp(
+    subq: EventsSubQuery,
+    node: Node,
+) -> Result<Pin<Box<dyn Stream<Item = Sitemty<EventFull>> + Send>>, Error> {
+    let addr = format!("{}:{}", node.host, node.port_raw);
+    debug!("x_processed_event_blobs_stream_from_node  to: {addr}",);
+    let frame1 = make_node_command_frame(subq.clone())?;
+    let net = TcpStream::connect(addr.clone()).await?;
+    let (netin, mut netout) = net.into_split();
+    let item = sitem_data(frame1);
+    let buf = item.make_frame_dyn()?;
+    netout.write_all(&buf).await?;
+    let buf = make_term_frame()?;
+    netout.write_all(&buf).await?;
+    netout.flush().await?;
+    netout.forget();
+    let inp = TcpReadAsBytes::new(netin).map_err(sitem_err2_from_string);
+    let inp = Box::pin(inp) as BoxedBytesStream;
+    let frames = InMemoryFrameStream::new(inp, subq.inmem_bufcap());
+    let frames = frames.map_err(sitem_err2_from_string);
+    let frames = Box::pin(frames);
+    let items = EventsFromFrames::new(frames, addr);
+    Ok(Box::pin(items))
+}
+
+#[allow(unused)]
+async fn open_event_data_streams_tcp<T>(subq: EventsSubQuery, cluster: &Cluster) -> Result<Vec<BoxedStream<T>>, Error>
+where
+    // TODO group bounds in new trait
+    T: FrameTypeInnerStatic + DeserializeOwned + Send + Unpin + fmt::Debug + 'static,
+{
+    // TODO when unit tests established, change to async connect:
+    let frame1 = make_node_command_frame(subq.clone())?;
+    let mut streams = Vec::new();
+    for node in &cluster.nodes {
+        let addr = format!("{}:{}", node.host, node.port_raw);
+        debug!("open_tcp_streams  to: {addr}");
+        let net = TcpStream::connect(addr.clone()).await?;
+        let (netin, mut netout) = net.into_split();
+        let item = sitem_data(frame1.clone());
+        let buf = item.make_frame_dyn()?;
+        netout.write_all(&buf).await?;
+        let buf = make_term_frame()?;
+        netout.write_all(&buf).await?;
+        netout.flush().await?;
+        netout.forget();
+        // TODO for images, we need larger buffer capacity
+        let inp = TcpReadAsBytes::new(netin);
+        let inp = inp.map_err(sitem_err2_from_string);
+        let inp = Box::pin(inp) as BoxedBytesStream;
+        let frames = InMemoryFrameStream::new(inp, subq.inmem_bufcap());
+        let frames = frames.map_err(sitem_err2_from_string);
+        let frames = Box::pin(frames);
+        let stream = EventsFromFrames::<T>::new(frames, addr);
+        streams.push(Box::pin(stream) as _);
+    }
+    Ok(streams)
+}
+
+// Currently used only for the python data api3 protocol endpoint.
+// TODO merge with main method.
+pub async fn x_processed_event_blobs_stream_from_node(
+    subq: EventsSubQuery,
+    node: Node,
+    post: Box<dyn HttpSimplePost>,
+    ctx: ReqCtx,
+) -> Result<Pin<Box<dyn Stream<Item = Sitemty<EventFull>> + Send>>, Error> {
+    if true {
+        Ok(x_processed_event_blobs_stream_from_node_http(subq, node, post, &ctx).await?)
+    } else {
+        x_processed_event_blobs_stream_from_node_tcp(subq, node).await
+    }
+}

crates/streamio/src/tcpreadasbytes.rs

@@ -0,0 +1,49 @@
+use bytes::Bytes;
+use futures_util::Stream;
+use std::pin::Pin;
+use std::task::Context;
+use std::task::Poll;
+use tokio::io::AsyncRead;
+
+#[derive(Debug, thiserror::Error)]
+#[cstm(name = "TcpReadAsBytes")]
+pub enum Error {
+    IO(#[from] std::io::Error),
+}
+
+pub struct TcpReadAsBytes<INP> {
+    inp: INP,
+}
+
+impl<INP> TcpReadAsBytes<INP> {
+    pub fn new(inp: INP) -> Self {
+        Self { inp }
+    }
+}
+
+impl<INP> Stream for TcpReadAsBytes<INP>
+where
+    INP: AsyncRead + Unpin,
+{
+    type Item = Result<Bytes, Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        use Poll::*;
+        let mut buf1 = vec![0; 512];
+        let mut buf2 = tokio::io::ReadBuf::new(&mut buf1);
+        match tokio::io::AsyncRead::poll_read(Pin::new(&mut self.inp), cx, &mut buf2) {
+            Ready(Ok(())) => {
+                let n = buf2.filled().len();
+                if n == 0 {
+                    Ready(None)
+                } else {
+                    buf1.truncate(n);
+                    let item = Bytes::from(buf1);
+                    Ready(Some(Ok(item)))
+                }
+            }
+            Ready(Err(e)) => Ready(Some(Err(e.into()))),
+            Pending => Pending,
+        }
+    }
+}