aj/bound event bus

bottlecap/src/event_bus/bus.rs

@@ -1,20 +1,21 @@
-use std::sync::mpsc;
-use std::sync::mpsc::Sender;
+use std::sync::mpsc::{self, SyncSender};
 
 use crate::events;
 
+use crate::event_bus::constants::MAX_EVENTS;
+
 pub struct EventBus {
-    tx: Sender<events::Event>,
+    tx: SyncSender<events::Event>,
     pub rx: mpsc::Receiver<events::Event>,
 }
 
 impl EventBus {
     pub fn run() -> EventBus {
-        let (tx, rx) = mpsc::channel();
+        let (tx, rx) = mpsc::sync_channel(MAX_EVENTS);
         EventBus { tx, rx }
     }
 
-    pub fn get_sender_copy(&self) -> Sender<events::Event> {
+    pub fn get_sender_copy(&self) -> SyncSender<events::Event> {
         self.tx.clone()
     }
 }

bottlecap/src/event_bus/constants.rs

@@ -0,0 +1 @@
+pub const MAX_EVENTS: usize = 100;

bottlecap/src/event_bus/mod.rs

@@ -1 +1,2 @@
 pub mod bus;
+mod constants;

bottlecap/src/main.rs

@@ -134,8 +134,10 @@ fn build_function_arn(account_id: &str, region: &str, function_name: &str) -> St
 
 fn main() -> Result<()> {
     // First load the configuration
-    let lambda_directory = env::var("LAMBDA_TASK_ROOT")
-        .expect("unable to read environment variable: LAMBDA_TASK_ROOT");
+    let lambda_directory = match env::var("LAMBDA_TASK_ROOT") {
+        Ok(val) => val,
+        Err(_) => "/var/task".to_string(),
+    };
     let config = match config::get_config(Path::new(&lambda_directory)) {
         Ok(config) => Arc::new(config),
         Err(e) => {

bottlecap/src/metrics/dogstatsd.rs

@@ -1,4 +1,4 @@
-use std::sync::mpsc::Sender;
+use std::sync::mpsc::SyncSender;
 
 use tracing::{error, info};
 
@@ -24,7 +24,7 @@ pub struct DogStatsDConfig {
 }
 
 impl DogStatsD {
-    pub fn run(config: &DogStatsDConfig, event_bus: Sender<events::Event>) -> DogStatsD {
+    pub fn run(config: &DogStatsDConfig, event_bus: SyncSender<events::Event>) -> DogStatsD {
         let aggr: Arc<Mutex<Aggregator<{ constants::CONTEXTS }>>> = Arc::new(Mutex::new(
             Aggregator::<{ constants::CONTEXTS }>::new().expect("failed to create aggregator"),
         ));
@@ -45,7 +45,7 @@ impl DogStatsD {
     fn run_server(
         host: &str,
         port: u16,
-        event_bus: Sender<events::Event>,
+        event_bus: SyncSender<events::Event>,
         aggregator: Arc<Mutex<Aggregator<{ constants::CONTEXTS }>>>,
     ) -> std::thread::JoinHandle<()> {
         let addr = format!("{}:{}", host, port);

bottlecap/src/telemetry/listener.rs

@@ -3,7 +3,7 @@ use crate::telemetry::events::TelemetryEvent;
 
 use std::collections::HashMap;
 use std::error::Error;
-use std::sync::mpsc::Sender;
+use std::sync::mpsc::SyncSender;
 use std::{
     io::{Read, Write},
     net::{TcpListener, TcpStream},
@@ -118,7 +118,7 @@ pub struct TelemetryListenerConfig {
 impl TelemetryListener {
     pub fn run(
         config: &TelemetryListenerConfig,
-        event_bus: Sender<events::Event>,
+        event_bus: SyncSender<events::Event>,
     ) -> Result<TelemetryListener, Box<dyn Error>> {
         let addr = format!("{}:{}", &config.host, &config.port);
         let listener = TcpListener::bind(addr)?;
@@ -152,7 +152,7 @@ impl TelemetryListener {
     fn handle_stream(
         stream: &mut impl Read,
         mut buf: [u8; 262144],
-        event_bus: Sender<events::Event>,
+        event_bus: SyncSender<events::Event>,
     ) -> Result<(), Box<dyn Error>> {
         // Read into buffer
         #![allow(clippy::unused_io_amount)]
@@ -294,7 +294,7 @@ mod tests {
         let mut stream = MockTcpStream {
             data: "POST /path HTTP/1.1\r\nContent-Length: 335\r\nHeader1: Value1\r\n\r\n[{\"time\":\"2024-04-25T17:35:59.944Z\",\"type\":\"platform.initStart\",\"record\":{\"initializationType\":\"on-demand\",\"phase\":\"init\",\"runtimeVersion\":\"nodejs:20.v22\",\"runtimeVersionArn\":\"arn:aws:lambda:us-east-1::runtime:da57c20c4b965d5b75540f6865a35fc8030358e33ec44ecfed33e90901a27a72\",\"functionName\":\"hello-world\",\"functionVersion\":\"$LATEST\"}}]".to_string().into_bytes(),
         };
-        let (tx, rx) = std::sync::mpsc::channel();
+        let (tx, rx) = std::sync::mpsc::sync_channel(3);
         let buf = [0; 262144];
         let result = TelemetryListener::handle_stream(&mut stream, buf, tx);
         let event = rx.recv().expect("No events received");
@@ -323,7 +323,7 @@ mod tests {
                     let mut stream = MockTcpStream {
                         data: $value.to_string().into_bytes(),
                     };
-                    let (tx, _) = std::sync::mpsc::channel();
+                    let (tx, _) = std::sync::mpsc::sync_channel(4);
                     let buf = [0; 262144];
                     TelemetryListener::handle_stream(&mut stream, buf, tx).unwrap()
                 }