339                     identifier,  in event_loop()
385                         .publish(identifier.clone(), receiver, stream_handler)  in event_loop()
393                                 .insert(info.id, PubSubInfo::Publish { identifier });  in event_loop()
409                     identifier,  in event_loop()
412                     if let Err(err) = self.unpublish(&identifier) {  in event_loop()
416                             identifier  in event_loop()
425                     identifier,  in event_loop()
460                     let rv = match self.subscribe(&identifier, info_clone, sender).await {  in event_loop()
469                                     identifier,  in event_loop()
485                 StreamHubEvent::UnSubscribe { identifier, info } => {  in event_loop()
486                     if self.unsubscribe(&identifier, info).is_ok() {  in event_loop()
508                 StreamHubEvent::Request { identifier, sender } => {  in event_loop()
509                     if let Err(err) = self.request(&identifier, sender) {  in event_loop()
519         identifier: &StreamIdentifier,  in request()
522         if let Some(producer) = self.streams.get_mut(identifier) {  in request()
524             log::info!("Request:  stream identifier: {}", identifier);  in request()
568             PubSubInfo::Publish { identifier } => {  in api_kick_off_client()
569                 if let Err(err) = self.unpublish(&identifier) {  in api_kick_off_client()
573                         identifier  in api_kick_off_client()
578                 identifier,  in api_kick_off_client()
581                 if let Err(err) = self.unsubscribe(&identifier, sub_info) {  in api_kick_off_client()
585                         identifier  in api_kick_off_client()
616                 identifier: identifer.clone(),  in subscribe()
658         identifier: StreamIdentifier,  in publish()
662         if self.streams.get(&identifier).is_some() {  in publish()
671         let identifier_clone = identifier.clone();  in publish()
683         self.streams.insert(identifier.clone(), event_publisher);  in publish()
686             let client_event = BroadcastEvent::Publish { identifier };  in publish()
699     fn unpublish(&mut self, identifier: &StreamIdentifier) -> Result<(), ChannelError> {  in unpublish()
700         match self.streams.get_mut(identifier) {  in unpublish()
706                 self.streams.remove(identifier);  in unpublish()
707                 log::info!("unpublish remove stream, stream identifier: {}", identifier);  in unpublish()