[Brook K8] Served reads with long polls, Topic filters, last-N reads and byte caps.

crates/streams/src/lib.rs

@@ -29,6 +29,7 @@ pub use runtime::{
     MEMORY_RUNTIME_NAME, MemoryStreamRuntime, REDIS_RUNTIME_NAME, RedisStreamRuntime,
     RuntimeOptions, StreamRuntime,
 };
+pub use service::{MAX_RECORDS, MAX_WAIT};
 
 /// The protos that lang and the crate exchange.
 pub mod proto {

crates/streams/src/metrics.rs

@@ -1,14 +1,15 @@
 //! The crate's metrics. They use the meter that lang gives, or a no-op meter.
 
 use temporalio_common::telemetry::metrics::{
-    Counter, MetricAttributes, MetricParameters, TemporalMeter,
+    Counter, MetricAttributes, MetricParameters, TemporalMeter, UpDownCounter,
 };
 
 #[derive(Clone)]
 pub(crate) struct Metrics {
     attributes: MetricAttributes,
     fenced: Counter,
     superseded: Counter,
+    blocking_reads_waiting: UpDownCounter,
 }
 
 fn parameters(
@@ -37,6 +38,11 @@ impl Metrics {
                 "Superseded flags that the Store set",
                 "",
             )),
+            blocking_reads_waiting: meter.up_down_counter(parameters(
+                "stream_blocking_reads_waiting",
+                "Reads that wait on the Store",
+                "",
+            )),
         }
     }
 
@@ -47,4 +53,9 @@ impl Metrics {
     pub(crate) fn superseded(&self) {
         self.superseded.add(1, &self.attributes);
     }
+
+    pub(crate) fn read_waits(&self, waiting: bool) {
+        self.blocking_reads_waiting
+            .add(if waiting { 1 } else { -1 }, &self.attributes);
+    }
 }

crates/streams/src/runtime.rs

@@ -4,7 +4,10 @@ use crate::{
     Reason, StreamError, StreamResult,
     memory::MemoryStore,
     metrics::Metrics,
-    proto::{AppendRecordsRequest, CloseStreamRequest, DescribeStreamRequest, RedisStreamConfig},
+    proto::{
+        AppendRecordsRequest, CloseStreamRequest, DescribeStreamRequest, ReadRecordsRequest,
+        RedisStreamConfig,
+    },
     redis::{RedisStore, RedisStoreOptions},
     service::StreamService,
     store::Store,
@@ -173,6 +176,10 @@ impl StreamRuntime {
                 .close_stream(decode::<CloseStreamRequest>(input)?)
                 .await?
                 .encode_to_vec(),
+            "ReadRecords" => service
+                .read_records(decode::<ReadRecordsRequest>(input)?)
+                .await?
+                .encode_to_vec(),
             "DescribeStream" => service
                 .describe_stream(decode::<DescribeStreamRequest>(input)?)
                 .await?