From 075bcc369f23b4f75a41efe6dbafad6dad6b87b2 Mon Sep 17 00:00:00 2001
From: 0xKitsune <0xKitsune@protonmail.com>
Date: Wed, 31 Jan 2024 22:21:23 -0500
Subject: [PATCH] updated stream connection

---
 bin/mpc_participant.rs |  8 ++++++++
 src/participant.rs     | 16 ++++++++++------
 2 files changed, 18 insertions(+), 6 deletions(-)

diff --git a/bin/mpc_participant.rs b/bin/mpc_participant.rs
index b383f4d..d36f22f 100644
--- a/bin/mpc_participant.rs
+++ b/bin/mpc_participant.rs
@@ -1,6 +1,8 @@
+use std::net::{SocketAddr, SocketAddrV4};
 use std::path::PathBuf;
 
 use clap::Parser;
+use mpc::participant::Participant;
 use telemetry_batteries::metrics::batteries::StatsdBattery;
 use telemetry_batteries::tracing::batteries::DatadogBattery;
 use tracing_subscriber::layer::SubscriberExt;
@@ -59,5 +61,11 @@ async fn main() -> eyre::Result<()> {
         .add_source(config::Environment::with_prefix("MPC").separator("__"))
         .build()?;
 
+    let participant =
+        Participant::new("127.0.0.1:8080".parse::<SocketAddr>()?, 20_000)
+            .await?;
+
+    participant.spawn().await?;
+
     Ok(())
 }
diff --git a/src/participant.rs b/src/participant.rs
index c190be2..e6cb866 100644
--- a/src/participant.rs
+++ b/src/participant.rs
@@ -8,20 +8,25 @@ use crate::distance::{self, DistanceEngine, EncodedBits};
 
 pub struct Participant {
     listener: tokio::net::TcpListener,
+    batch_size: usize,
 }
 
 impl Participant {
-    pub async fn new(socket_address: SocketAddr) -> eyre::Result<Self> {
+    pub async fn new(
+        socket_address: SocketAddr,
+        batch_size: usize,
+    ) -> eyre::Result<Self> {
         Ok(Self {
             listener: tokio::net::TcpListener::bind(socket_address).await?,
+            batch_size,
         })
     }
 
     pub async fn spawn(&self) -> eyre::Result<()> {
-        let (stream, _) = self.listener.accept().await?;
-        let mut stream = tokio::io::BufWriter::new(stream);
-
+        let mut stream =
+            tokio::io::BufWriter::new(self.listener.accept().await?.0);
         let shares = Arc::new(self.initialize_shares().await?);
+        let batch_size = self.batch_size;
 
         loop {
             // TODO: Sync from database
@@ -39,8 +44,7 @@ impl Participant {
                     bytemuck::cast_slice(&shares_ref);
                 let engine = DistanceEngine::new(&distance::encode(&template));
 
-                //TODO: make batch size configurable
-                for chunk in patterns.chunks(20_000) {
+                for chunk in patterns.chunks(batch_size) {
                     let mut result = vec![
                         0_u8;
                         chunk.len()