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 println!("push client run..."); 32 let mut session_id = std::u64::MAX; 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 println!( 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 session_id, 55 ); 56 57 tokio::spawn(async move { 58 if let Err(err) = client_session.run().await { 59 print!(" session error {}\n", err); 60 } 61 }); 62 63 session_id = session_id - 1; 64 } 65 66 _ => { 67 println!("other infos..."); 68 } 69 } 70 } 71 } 72 } 73