1 use { 2 super::errors::ClientError, 3 crate::{ 4 channels::define::{ChannelEventProducer, ClientEvent, ClientEventConsumer}, 5 session::client_session::{ClientSession, ClientType}, 6 }, 7 tokio::net::TcpStream, 8 }; 9 10 pub struct PullClient { 11 address: String, 12 client_event_consumer: ClientEventConsumer, 13 channel_event_producer: ChannelEventProducer, 14 } 15 16 impl PullClient { 17 pub fn new( 18 address: String, 19 consumer: ClientEventConsumer, 20 producer: ChannelEventProducer, 21 ) -> Self { 22 Self { 23 address: address, 24 25 client_event_consumer: consumer, 26 channel_event_producer: producer, 27 } 28 } 29 30 pub async fn run(&mut self) -> Result<(), ClientError> { 31 let mut session_id = std::u64::MAX; 32 33 loop { 34 let val = self.client_event_consumer.recv().await?; 35 match val { 36 ClientEvent::Subscribe { 37 app_name, 38 stream_name, 39 } => { 40 let stream = TcpStream::connect(self.address.clone()).await?; 41 42 let mut client_session = ClientSession::new( 43 stream, 44 ClientType::Play, 45 app_name.clone(), 46 stream_name.clone(), 47 self.channel_event_producer.clone(), 48 session_id, 49 ); 50 51 tokio::spawn(async move { 52 if let Err(err) = client_session.run().await { 53 print!(" session error {}\n", err); 54 } 55 }); 56 57 session_id = session_id - 1; 58 } 59 _ => {} 60 } 61 } 62 } 63 } 64