async fn connection_loop(client_id: i32, broker: Sender<ClientEvent>, stream: TcpStream) -> Result<()> {
 let (socket_reader, mut socket_writer) = (&stream, &stream);
 let reader = BufReader::new(socket_reader);
 let mut lines = reader.lines();
 let name = match lines.next().await {
 None => Err("peer disconnected immediately")?,
 Some(line) => line?,
 };
 broker.send(Event::NewPeer { name: name.clone(), stream: Arc::clone(&stream) }).await // 3
 .unwrap();
 let (peer_sender, peer_receiver) = channel::unbounded::<String>();
 broker
 .send(ClientEvent::NewPeer {
 client_id,
 sender: peer_sender,
 })
 .await
 .unwrap();
 loop {
 select! {
 line = lines.next().fuse() => {
 let line = line?;
 let (dest, msg) = match line.find(':') {
 None => continue,
 Some(idx) => (&line[..idx], line[idx + 1 ..].trim()),
 };
 let dest: Vec<String> = dest.split(',').map(|name| name.trim().to_string()).collect();
 let msg: String = msg.to_string();
 broker.send(Event::Message { // 4
 from: name.clone(),
 to: dest,
 msg,
 }).await.unwrap();
 },
 message = peer_receiver.next().fuse() => match message {
 None => {
 break;
 }
 Some(message) => {
 socket_writer.send(msg.as_bytes()).await?;
 }
 }
 }
 }
 broker.send(ClientEvent::PeerGone { client_id }).await.unwrap();
 Ok(())
}