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 PushClient { 11 address: String, 12 client_event_consumer: ClientEventConsumer, 13 channel_event_producer: ChannelEventProducer, 14 } 15 16 impl PushClient { 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 log::info!("push client run..."); 32 33 loop { 34 let val = self.client_event_consumer.recv().await?; 35 match val { 36 ClientEvent::Publish { 37 app_name, 38 stream_name, 39 } => { 40 log::info!( 41 "publish app_name: {} stream_name: {} address: {}", 42 app_name.clone(), 43 stream_name.clone(), 44 self.address.clone() 45 ); 46 let stream = TcpStream::connect(self.address.clone()).await?; 47 48 let mut client_session = ClientSession::new( 49 stream, 50 ClientType::Publish, 51 app_name.clone(), 52 stream_name.clone(), 53 self.channel_event_producer.clone(), 54 ); 55 56 tokio::spawn(async move { 57 if let Err(err) = client_session.run().await { 58 log::error!("client_session as push client run error: {}", err); 59 } 60 }); 61 } 62 63 _ => { 64 log::info!("push client receive other events"); 65 } 66 } 67 } 68 } 69 } 70