You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: portal_worker.rs
    path of file: ./repos/ockam/implementations/rust/ockam/ockam_transport_tcp/src/portal
    
    the code of the file until where you have to start completion: use crate::portal::addresses::{Addresses, PortalType};
use crate::{portal::TcpPortalRecvProcessor, PortalInternalMessage, PortalMessage, TcpRegistry};
use
    
    async fn handle_message(&mut self, ctx: &mut Context, msg: Routed<Any>) -> Result<()> {
        if self.is_disconnecting {
            return Ok(());
        }

        // Remove our own address from the route so the other end
        // knows what to do with the incoming message
        let state = self.clone_state();
        let mut onward_route = msg.onward_route();
        let recipient = onward_route.step()?;
        if onward_route.next().is_ok() {
            return Err(TransportError::UnknownRoute)?;
        }
        let return_route = msg.return_route();
        let remote_packet = recipient != self.addresses.internal;

        match state {
            State::ReceivePong => {
                if !remote_packet {
                    return Err(TransportError::PortalInvalidState)?;
                };
                if PortalMessage::decode(msg.payload())? != PortalMessage::Pong {
                    return Err(TransportError::Protocol)?;
                };
                self.handle_receive_pong(ctx, return_route).await
            }
            State::Initialized => {
                trace!(
                    "{:?} at: {} received {} tcp packet",
                    self.portal_type.str(),
                    self.addresses.internal,
                    if remote_packet { "remote" } else { "internal " }
                );

                if remote_packet {
                    let msg = PortalMessage::decode(msg.payload())?;
                    // Send to Tcp stream
                    match msg {
                        PortalMessage::Payload(payload, packet_counter) => {
                            self.handle_payload(ctx, payload, packet_counter).await
                        }
                        PortalMessage::Disconnect => {
                            self.start_disconnection(ctx, DisconnectionReason::Remote)
                                .await
                        }
                        PortalMessage::Ping | PortalMessage::Pong => {
                            return Err(TransportError::Protocol)?;
                        }
                    }
                } else {
                    let msg = PortalInternalMessage::decode(msg.payload())?;
                    if msg != PortalInternalMessage::Disconnect {
                        return Err(TransportError::Protocol)?;
                    };
                    self.handle_disconnect(ctx).await
                }
            }
            State::SendPing { .. } | State::SendPong { .. } => {
                return Err(TransportError::PortalInvalidState)?
            }
        }
    }