wip: idk about this

kit/src/behaviors/swap/mod.rs

@@ -16,6 +16,28 @@ where
     pub _phantom: PhantomData<E>,
 }
 
+pub trait SwapStream<SwapType, E> where E: Send + 'static {
+    fn get_typed_stream(swap_type: SwapType, channel: Messager, client: Arc<ArbiterMiddleware>) -> Result<Option<EventStream<E>>>;
+}
+
+
+#[derive(Deserialize, Clone)]
+pub struct SwapOnce {
+    pub amount: eU256,
+    pub input: InputToken,
+}
+
+
+impl<P, SwapOnce, Message> SwapStream<SwapOnce, Message> for Swap<Processing<P>, SwapOnce, Message>
+where
+    P: PoolType + Send + Sync,
+{
+    fn get_typed_stream(swap_type: SwapOnce, channel: Messager, client: Arc<ArbiterMiddleware>) -> Result<Option<EventStream<Message>>> {
+        let thing = channel.stream()?;
+        Ok(Some(thing))
+    }
+}
+
 // TODO: This needs to be configurable in some way to make the `SwapType` become
 // transparent and useful.
 // Should also get some data necessary for mint amounts and what not.
@@ -46,7 +68,7 @@ where
 impl<P, T, E> Behavior<E> for Swap<Config<P>, T, E>
 where
     P: PoolType + Send + Sync,
-    T: SwapType<E> + Send,
+    T: SwapType<E> + Send + Clone,
     E: Send + 'static,
 {
     type Processor = Swap<Processing<P>, T, E>;
@@ -125,11 +147,12 @@ where
 impl<P, T, E> Processor<E> for Swap<Processing<P>, T, E>
 where
     P: PoolType + Send + Sync,
-    T: SwapType<E> + Send,
+    T: SwapType<E> + Send + Clone,
     E: Send + 'static,
+    Swap<Processing<P>, T, E>: SwapStream<T, E>,
 {
     async fn get_stream(&mut self) -> Result<Option<EventStream<E>>> {
-        todo!("We have not implemented the 'get_stream' method yet for the 'Swap' behavior.")
+        behaviors::swap::Swap::<behaviors::swap::Processing<P>, T, E>::get_typed_stream(self.swap_type.clone(), self.data.messager.clone(), self.data.client)
     }
     async fn process(&mut self, event: E) -> Result<ControlFlow> {
         let (swap_amount, input) = self.swap_type.compute_swap_amount(event);

kit/tests/common.rs

@@ -1,23 +1,26 @@
 use std::{collections::VecDeque, marker::PhantomData};
 
-use arbiter_engine::{agent::Agent, messager::Message, world::World};
+use arbiter_core::middleware::ArbiterMiddleware;
+use arbiter_engine::{agent::Agent, machine::State, messager::{Message, Messager}, world::World};
 use dfmm_kit::{
     behaviors::{
         creator::{self, Create},
         deploy::Deploy,
-        swap::{self, Swap, SwapType},
+        swap::{self, Swap, SwapStream, SwapType},
         token::{self, TokenAdmin},
         update::{self, Update},
     },
     bindings::constant_sum_solver::ConstantSumParams,
     pool::{
         constant_sum::{ConstantSumAllocationData, ConstantSumPool},
-        BaseConfig, InputToken, PoolCreation,
+        BaseConfig, InputToken,
     },
     TokenData,
 };
+
+use anyhow::Result;
 use ethers::types::{Address as eAddress, U256 as eU256};
-use serde::{Deserialize, Serialize};
+use serde::Deserialize;
 use tracing::Level;
 use tracing_subscriber::FmtSubscriber;
 
@@ -106,6 +109,12 @@ impl SwapType<Message> for SwapOnce {
         (self.amount, self.input.clone())
     }
 }
+impl<S> SwapStream<SwapOnce, Message> for Swap<S, SwapOnce, Message> where S: State{
+    fn get_typed_stream(_swap_type: SwapOnce, mut channel: Messager, _client: Arc<ArbiterMiddleware>) -> Result<Option<arbiter_engine::machine::EventStream<Message>>> {
+        let thing = channel.stream()?;
+        Ok(Some(thing))
+    }
+}
 
 fn mock_token_admin_behavior() -> TokenAdmin<token::Config> {
     TokenAdmin::<token::Config> {