<commit_before>use crate::{
    acknowledgement::Acknowledgements,
    auth::Credentials,
    basic_get_delivery::BasicGetDelivery,
    channel_closer::ChannelCloser,
    channel_receiver_state::DeliveryCause,
    channel_status::{ChannelState, ChannelStatus},
    connection_closer::ConnectionCloser,
    connection_status::{ConnectionState, ConnectionStep},
    consumer::Consumer,
    consumers::Consumers,
    executor::Executor,
    frames::{ExpectedReply, Frames},
    internal_rpc::InternalRPCHandle,
    message::{BasicGetMessage, BasicReturnMessage, Delivery},
    protocol::{self, AMQPClass, AMQPError, AMQPHardError},
    publisher_confirm::PublisherConfirm,
    queue::Queue,
    registry::Registry,
    returned_messages::ReturnedMessages,
    socket_state::SocketStateHandle,
    topology::RestoredChannel,
    topology_internal::ChannelDefinitionInternal,
    types::*,
    BasicProperties, Configuration, ConfirmationPromise, Connection, ConnectionStatus, DeliveryTag,
    Error, ExchangeKind, Promise, PromiseChain, PromiseResolver, Result,
};
use amq_protocol::frame::{AMQPContentHeader, AMQPFrame};
use log::{debug, error, info, log_enabled, trace, Level::Trace};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::{convert::TryFrom, fmt, sync::Arc};

#[derive(Clone)]
pub struct Channel {
    id: u16,
    configuration: Configuration,
    status: ChannelStatus,
    connection_status: ConnectionStatus,
    global_registry: Registry,
    local_registry: Registry,
    acknowledgements: Acknowledgements,
    consumers: Consumers,
    basic_get_delivery: BasicGetDelivery,
    returned_messages: ReturnedMessages,
    waker: SocketStateHandle,
    internal_rpc: InternalRPCHandle,
    frames: Frames,
    executor: Arc<dyn Executor>,
    channel_closer: Option<Arc<ChannelCloser>>,
    connection_closer: Option<Arc<ConnectionCloser>>,
}

impl PartialEq for Channel {
    fn eq(&self, other: &Self) -> bool {
        self.id == other.id
    }
}

impl fmt::Debug for Channel {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Channel")
            .field("id", &self.id)
            .field("configuration", &self.configuration)
            .field("status", &self.status)
            .field("connection_status", &self.connection_status)
            .field("acknowledgements", &self.acknowledgements)
            .field("consumers", &self.consumers)
            .field("basic_get_delivery", &self.basic_get_delivery)
            .field("returned_messages", &self.returned_messages)
            .field("frames", &self.frames)
            .field("executor", &self.executor)
            .finish()
    }
}

impl Channel {
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        channel_id: u16,
        configuration: Configuration,
        connection_status: ConnectionStatus,
        global_registry: Registry,
        waker: SocketStateHandle,
        internal_rpc: InternalRPCHandle,
        frames: Frames,
        executor: Arc<dyn Executor>,
        connection_closer: Option<Arc<ConnectionCloser>>,
    ) -> Channel {
        let returned_messages = ReturnedMessages::default();
        let status = ChannelStatus::default();
        let channel_closer = if channel_id == 0 {
            None
        } else {
            Some(Arc::new(ChannelCloser::new(
                channel_id,
                status.clone(),
                internal_rpc.clone(),
            )))
        };
        Channel {
            id: channel_id,
            configuration,
            status,
            connection_status,
            global_registry,
            local_registry: Registry::default(),
            acknowledgements: Acknowledgements::new(channel_id, returned_messages.clone()),
            consumers: Consumers::default(),
            basic_get_delivery: BasicGetDelivery::default(),
            returned_messages,
            waker,
            internal_rpc,
            frames,
            executor,
            channel_closer,
            connection_closer,
        }
    }

    pub fn status(&self) -> &ChannelStatus {
        &self.status
    }

    pub(crate) fn reset(&self) {
        // FIXME
    }

    pub(crate) async fn restore(
        &self,
        ch: &ChannelDefinitionInternal,
        c: &mut RestoredChannel,
    ) -> Result<()> {
        // First, redeclare all queues
        for queue in &ch.queues {
            if queue.is_declared() {
                c.queues.push(
                    self.queue_declare(
                        queue.name.as_str(),
                        queue.options.unwrap_or_default(),
                        queue.arguments.clone().unwrap_or_default(),
                    )
                    .await?,
                );
            }
        }

        // Second, redeclare all queues bindings
        for queue in &ch.queues {
            for binding in &queue.bindings {
                self.queue_bind(
                    queue.name.as_str(),
                    binding.source.as_str(),
                    binding.routing_key.as_str(),
                    QueueBindOptions::default(),
                    binding.arguments.clone(),
                )
                .await?;
            }
        }

        // Third, redeclare all consumers
        for consumer in &ch.consumers {
            let original = consumer.original();
            if let Some(original) = original.as_ref() {
                original.reset();
            }
            c.consumers.push(
                self.do_basic_consume(
                    consumer.queue.as_str(),
                    consumer.tag.as_str(),
                    consumer.options,
                    consumer.arguments.clone(),
                    original,
                )
                .await?,
            );
        }

        // Fourth, reemit pending basic_get
        if let Some(original) = self.basic_get_delivery.recover() {
            self.do_basic_get(
                original.queue.as_str(),
                original.options,
                Some(original.resolver),
            )
            .await?;
        }

        Ok(())
    }

    fn set_closing(&self) {
        self.set_state(ChannelState::Closing);
        self.consumers.start_cancel();
    }

    fn set_closed(&self, error: Error) -> Result<()> {
        self.set_state(ChannelState::Closed);
        self.error_publisher_confirms(error.clone());
        let res = self.cancel_consumers();
        self.internal_rpc.remove_channel(self.id, error);
        res
    }

    fn set_error(&self, error: Error) -> Result<()> {
        self.set_state(ChannelState::Error);
        self.error_publisher_confirms(error.clone());
        let res = self.error_consumers(error.clone());
        self.internal_rpc.remove_channel(self.id, error);
        res
    }

    pub(crate) fn error_publisher_confirms(&self, error: Error) {
        self.acknowledgements.on_channel_error(error);
    }

    pub(crate) fn cancel_consumers(&self) -> Result<()> {
        self.consumers.cancel()
    }

    pub(crate) fn error_consumers(&self, error: Error) -> Result<()> {
        self.consumers.error(error)
    }

    pub(crate) fn set_state(&self, state: ChannelState) {
        self.status.set_state(state);
    }

    pub fn id(&self) -> u16 {
        self.id
    }

    pub(crate) fn clone_internal(&self) -> Self {
        Self {
            id: self.id,
            configuration: self.configuration.clone(),
            status: self.status.clone(),
            connection_status: self.connection_status.clone(),
            global_registry: self.global_registry.clone(),
            local_registry: self.local_registry.clone(),
            acknowledgements: self.acknowledgements.clone(),
            consumers: self.consumers.clone(),
            basic_get_delivery: self.basic_get_delivery.clone(),
            returned_messages: self.returned_messages.clone(),
            waker: self.waker.clone(),
            internal_rpc: self.internal_rpc.clone(),
            frames: self.frames.clone(),
            executor: self.executor.clone(),
            channel_closer: None,
            connection_closer: self.connection_closer.clone(),
        }
    }

    fn wake(&self) {
        trace!("channel {} wake", self.id);
        self.waker.wake()
    }

    fn assert_channel0(&self, class_id: u16, method_id: u16) -> Result<()> {
        if self.id == 0 {
            Ok(())
        } else {
            error!(
                "Got a connection frame on channel {}, closing connection",
                self.id
            );
            let error = AMQPError::new(
                AMQPHardError::COMMANDINVALID.into(),
                format!("connection frame received on channel {}", self.id).into(),
            );
            self.internal_rpc.close_connection(
                error.get_id(),
                error.get_message().to_string(),
                class_id,
                method_id,
            );
            Err(Error::ProtocolError(error))
        }
    }

    pub fn close(&self, reply_code: ShortUInt, reply_text: &str) -> Promise<()> {
        self.do_channel_close(reply_code, reply_text, 0, 0)
    }

    pub fn basic_consume(
        &self,
        queue: &str,
        consumer_tag: &str,
        options: BasicConsumeOptions,
        arguments: FieldTable,
    ) -> PromiseChain<Consumer> {
        self.do_basic_consume(queue, consumer_tag, options, arguments, None)
    }

    pub fn basic_get(
        &self,
        queue: &str,
        options: BasicGetOptions,
    ) -> PromiseChain<Option<BasicGetMessage>> {
        self.do_basic_get(queue, options, None)
    }

    pub fn exchange_declare(
        &self,
        exchange: &str,
        kind: ExchangeKind,
        options: ExchangeDeclareOptions,
        arguments: FieldTable,
    ) -> Promise<()> {
        self.do_exchange_declare(exchange, kind.kind(), options, arguments, kind.clone())
    }

    pub fn wait_for_confirms(&self) -> ConfirmationPromise<Vec<BasicReturnMessage>> {
        if let Some(promise) = self.acknowledgements.get_last_pending() {
            trace!("Waiting for pending confirms");
            let returned_messages = self.returned_messages.clone();
            promise.traverse(move |_| Ok(returned_messages.drain()))
        } else {
            trace!("No confirms to wait for");
            ConfirmationPromise::new_with_data(Ok(Vec::default()))
        }
    }

    #[cfg(test)]
    pub(crate) fn register_queue(
        &self,
        name: ShortString,
        options: QueueDeclareOptions,
        arguments: FieldTable,
    ) {
        self.local_registry.register_queue(name, options, arguments);
    }

    #[cfg(test)]
    pub(crate) fn register_consumer(&self, tag: ShortString, consumer: Consumer) {
        self.consumers.register(tag, consumer);
    }

    pub(crate) fn send_method_frame(
        &self,
        method: AMQPClass,
        resolver: PromiseResolver<()>,
        expected_reply: Option<ExpectedReply>,
    ) {
        self.send_frame(AMQPFrame::Method(self.id, method), resolver, expected_reply);
    }

    pub(crate) fn send_frame(
        &self,
        frame: AMQPFrame,
        resolver: PromiseResolver<()>,
        expected_reply: Option<ExpectedReply>,
    ) {
        trace!("channel {} send_frame", self.id);
        self.frames.push(self.id, frame, resolver, expected_reply);
        self.wake();
    }

    fn send_method_frame_with_body(
        &self,
        method: AMQPClass,
        payload: Vec<u8>,
        properties: BasicProperties,
        publisher_confirms_result: Option<PublisherConfirm>,
    ) -> Result<PromiseChain<PublisherConfirm>> {
        let class_id = method.get_amqp_class_id();
        let header = AMQPContentHeader {
            class_id,
            weight: 0,
            body_size: payload.len() as u64,
            properties,
        };
        let frame_max = self.configuration.frame_max();
        let mut frames = vec![
            AMQPFrame::Method(self.id, method),
            AMQPFrame::Header(self.id, class_id, Box::new(header)),
        ];

        // a content body frame 8 bytes of overhead
        frames.extend(
            payload
                .as_slice()
                .chunks(frame_max as usize - 8)
                .map(|chunk| AMQPFrame::Body(self.id, chunk.into())),
        );

        // tweak to make rustc happy
        let data = Arc::new(Mutex::new((
            publisher_confirms_result,
            self.returned_messages.clone(),
        )));

        trace!("channel {} send_frames", self.id);
        let promise = self.frames.push_frames(frames);
        self.wake();
        Ok(promise.traverse(move |res| {
            res.map(|()| {
                let mut data = data.lock();
                data.0
                    .take()
                    .unwrap_or_else(|| PublisherConfirm::not_requested(data.1.clone()))
            })
        }))
    }

    fn handle_invalid_contents(&self, error: String, class_id: u16, method_id: u16) -> Result<()> {
        error!("{}", error);
        let error = AMQPError::new(AMQPHardError::UNEXPECTEDFRAME.into(), error.into());
        self.internal_rpc.close_connection(
            error.get_id(),
            error.get_message().to_string(),
            class_id,
            method_id,
        );
        Err(Error::ProtocolError(error))
    }

    pub(crate) fn handle_content_header_frame(
        &self,
        class_id: u16,
        size: LongLongUInt,
        properties: BasicProperties,
    ) -> Result<()> {
        self.status.set_content_length(
            self.id,
            class_id,
            size,
            |delivery_cause, confirm_mode| {
                match delivery_cause {
                    DeliveryCause::Consume(consumer_tag) => {
                        self.consumers.handle_content_header_frame(
                            &self,
                            consumer_tag,
                            size,
                            properties,
                        )?;
                    }
                    DeliveryCause::Get => {
                        self.basic_get_delivery
                            .handle_content_header_frame(size, properties);
                    }
                    DeliveryCause::Return => {
                        self.returned_messages.handle_content_header_frame(
                            size,
                            properties,
                            confirm_mode,
                        );
                    }
                }
                Ok(())
            },
            |msg| {
                error!("{}", msg);
                let error = AMQPError::new(AMQPHardError::FRAMEERROR.into(), msg.into());
                self.internal_rpc.close_connection(
                    error.get_id(),
                    error.get_message().to_string(),
                    class_id,
                    0,
                );
                self.set_error(Error::ProtocolError(error))
            },
            |msg| self.handle_invalid_contents(msg, class_id, 0),
        )
    }

    pub(crate) fn handle_body_frame(&self, payload: Vec<u8>) -> Result<()> {
        self.status.receive(
            self.id,
            payload.len() as LongLongUInt,
            |delivery_cause, remaining_size, confirm_mode| {
                match delivery_cause {
                    DeliveryCause::Consume(consumer_tag) => {
                        self.consumers.handle_body_frame(
                            &self,
                            consumer_tag,
                            remaining_size,
                            payload,
                        )?;
                    }
                    DeliveryCause::Get => {
                        self.basic_get_delivery
                            .handle_body_frame(remaining_size, payload);
                    }
                    DeliveryCause::Return => {
                        self.returned_messages.handle_body_frame(
                            remaining_size,
                            payload,
                            confirm_mode,
                        );
                    }
                }
                Ok(())
            },
            |msg| self.handle_invalid_contents(msg, 0, 0),
        )
    }

    pub(crate) fn topology(&self) -> ChannelDefinitionInternal {
        ChannelDefinitionInternal {
            channel: Some(self.clone()),
            queues: self.local_registry.queues_topology(true),
            consumers: self.consumers.topology(),
        }
    }

    fn before_basic_publish(&self) -> Option<PublisherConfirm> {
        if self.status.confirm() {
            Some(self.acknowledgements.register_pending())
        } else {
            None
        }
    }

    fn before_basic_cancel(&self, consumer_tag: &str) {
        self.consumers.start_cancel_one(consumer_tag);
    }

    fn acknowledgement_error(&self, error: AMQPError, class_id: u16, method_id: u16) -> Result<()> {
        error!("Got a bad acknowledgement from server, closing channel");
        self.internal_rpc
            .register_internal_future(self.do_channel_close(
                error.get_id(),
                error.get_message().as_str(),
                class_id,
                method_id,
            ))?;
        Err(Error::ProtocolError(error))
    }

    fn on_connection_start_ok_sent(
        &self,
        resolver: PromiseResolver<Connection>,
        connection: Connection,
        credentials: Credentials,
    ) -> Result<()> {
        self.connection_status
            .set_connection_step(ConnectionStep::StartOk(resolver, connection, credentials));
        Ok(())
    }

    fn on_connection_open_sent(&self, resolver: PromiseResolver<Connection>) -> Result<()> {
        self.connection_status
            .set_connection_step(ConnectionStep::Open(resolver));
        Ok(())
    }

    fn on_connection_close_ok_sent(&self, error: Error) -> Result<()> {
        if let Error::ProtocolError(_) = error {
            self.internal_rpc.set_connection_error(error);
        } else {
            self.internal_rpc.set_connection_closed(error);
        }
        Ok(())
    }

    fn next_expected_close_ok_reply(&self) -> Option<Reply> {
        self.frames
            .next_expected_close_ok_reply(self.id, Error::InvalidChannelState(ChannelState::Closed))
    }

    fn before_channel_close(&self) {
        self.set_closing();
    }

    fn on_channel_close_ok_sent(&self, error: Error) -> Result<()> {
        self.set_closed(error)
    }

    fn on_basic_recover_async_sent(&self) -> Result<()> {
        self.consumers.drop_prefetched_messages()
    }

    fn on_basic_ack_sent(&self, multiple: bool, delivery_tag: DeliveryTag) -> Result<()> {
        if multiple && delivery_tag == 0 {
            self.consumers.drop_prefetched_messages()
        } else {
            Ok(())
        }
    }

    fn on_basic_nack_sent(&self, multiple: bool, delivery_tag: DeliveryTag) -> Result<()> {
        if multiple && delivery_tag == 0 {
            self.consumers.drop_prefetched_messages()
        } else {
            Ok(())
        }
    }

    fn tune_connection_configuration(&self, channel_max: u16, frame_max: u32, heartbeat: u16) {
        // If we disable the heartbeat (0) but the server don't, follow it and enable it too
        // If both us and the server want heartbeat enabled, pick the lowest value.
        if self.configuration.heartbeat() == 0
            || (heartbeat != 0 && heartbeat < self.configuration.heartbeat())
        {
            self.configuration.set_heartbeat(heartbeat);
        }

        if channel_max != 0 {
            // 0 means we want to take the server's value
            // If both us and the server specified a channel_max, pick the lowest value.
            if self.configuration.channel_max() == 0
                || channel_max < self.configuration.channel_max()
            {
                self.configuration.set_channel_max(channel_max);
            }
        }
        if self.configuration.channel_max() == 0 {
            self.configuration.set_channel_max(u16::max_value());
        }

        if frame_max != 0 {
            // 0 means we want to take the server's value
            // If both us and the server specified a frame_max, pick the lowest value.
            if self.configuration.frame_max() == 0 || frame_max < self.configuration.frame_max() {
                self.configuration.set_frame_max(frame_max);
            }
        }
        if self.configuration.frame_max() == 0 {
            self.configuration.set_frame_max(u32::max_value());
        }
    }

    fn on_connection_start_received(&self, method: protocol::connection::Start) -> Result<()> {
        trace!("Server sent connection::Start: {:?}", method);
        let state = self.connection_status.state();
        if let (
            ConnectionState::Connecting,
            Some(ConnectionStep::ProtocolHeader(
                resolver,
                connection,
                credentials,
                mechanism,
                mut options,
            )),
        ) = (state.clone(), self.connection_status.connection_step())
        {
            let mechanism_str = mechanism.to_string();
            let locale = options.locale.clone();

            if !method
                .mechanisms
                .split_whitespace()
                .any(|m| m == mechanism_str)
            {
                error!("unsupported mechanism: {}", mechanism);
            }
            if !method.locales.split_whitespace().any(|l| l == locale) {
                error!("unsupported locale: {}", mechanism);
            }

            if !options.client_properties.contains_key("product")
                || !options.client_properties.contains_key("version")
            {
                options.client_properties.insert(
                    "product".into(),
                    AMQPValue::LongString(env!("CARGO_PKG_NAME").into()),
                );
                options.client_properties.insert(
                    "version".into(),
                    AMQPValue::LongString(env!("CARGO_PKG_VERSION").into()),
                );
            }

            options
                .client_properties
                .insert("platform".into(), AMQPValue::LongString("rust".into()));

            let mut capabilities = FieldTable::default();
            capabilities.insert("publisher_confirms".into(), true.into());
            capabilities.insert("exchange_exchange_bindings".into(), true.into());
            capabilities.insert("basic.nack".into(), true.into());
            capabilities.insert("consumer_cancel_notify".into(), true.into());
            capabilities.insert("connection.blocked".into(), true.into());
            capabilities.insert("consumer_priorities".into(), true.into());
            capabilities.insert("authentication_failure_close".into(), true.into());
            capabilities.insert("per_consumer_qos".into(), true.into());
            capabilities.insert("direct_reply_to".into(), true.into());

            options
                .client_properties
                .insert("capabilities".into(), AMQPValue::FieldTable(capabilities));

            self.internal_rpc
                .register_internal_future(self.connection_start_ok(
                    options.client_properties,
                    &mechanism_str,
                    &credentials.sasl_auth_string(mechanism),
                    &locale,
                    resolver,
                    connection,
                    credentials,
                ))
        } else {
            error!("Invalid state: {:?}", state);
            let error = Error::InvalidConnectionState(state);
            self.internal_rpc.set_connection_error(error.clone());
            Err(error)
        }
    }

    fn on_connection_secure_received(&self, method: protocol::connection::Secure) -> Result<()> {
        trace!("Server sent connection::Secure: {:?}", method);

        let state = self.connection_status.state();
        if let (ConnectionState::Connecting, Some(ConnectionStep::StartOk(.., credentials))) =
            (state.clone(), self.connection_status.connection_step())
        {
            self.internal_rpc.register_internal_future(
                self.connection_secure_ok(&credentials.rabbit_cr_demo_answer()),
            )
        } else {
            error!("Invalid state: {:?}", state);
            let error = Error::InvalidConnectionState(state);
            self.internal_rpc.set_connection_error(error.clone());
            Err(error)
        }
    }

    fn on_connection_tune_received(&self, method: protocol::connection::Tune) -> Result<()> {
        debug!("Server sent Connection::Tune: {:?}", method);

        let state = self.connection_status.state();
        if let (
            ConnectionState::Connecting,
            Some(ConnectionStep::StartOk(resolver, connection, _)),
        ) = (state.clone(), self.connection_status.connection_step())
        {
            self.tune_connection_configuration(
                method.channel_max,
                method.frame_max,
                method.heartbeat,
            );

            self.internal_rpc
                .register_internal_future(self.connection_tune_ok(
                    self.configuration.channel_max(),
                    self.configuration.frame_max(),
                    self.configuration.heartbeat(),
                ))?;
            self.internal_rpc
                .register_internal_future(self.connection_open(
                    &self.connection_status.vhost(),
                    connection,
                    resolver,
                ))
        } else {
            error!("Invalid state: {:?}", state);
            let error = Error::InvalidConnectionState(state);
            self.internal_rpc.set_connection_error(error.clone());
            Err(error)
        }
    }

    fn on_connection_open_ok_received(
        &self,
        _: protocol::connection::OpenOk,
        connection: Connection,
    ) -> Result<()> {
        let state = self.connection_status.state();
        if let (ConnectionState::Connecting, Some(ConnectionStep::Open(resolver))) =
            (state.clone(), self.connection_status.connection_step())
        {
            self.connection_status.set_state(ConnectionState::Connected);
            resolver.swear(Ok(connection));
            Ok(())
        } else {
            error!("Invalid state: {:?}", state);
            let error = Error::InvalidConnectionState(state);
            self.internal_rpc.set_connection_error(error.clone());
            Err(error)
        }
    }

    fn on_connection_close_received(&self, method: protocol::connection::Close) -> Result<()> {
        let error = AMQPError::try_from(method.clone())
            .map(|error| {
                error!(
                    "Connection closed on channel {} by {}:{} => {:?} => {}",
                    self.id, method.class_id, method.method_id, error, method.reply_text
                );
                Error::ProtocolError(error)
            })
            .unwrap_or_else(|error| {
                error!("{}", error);
                info!("Connection closed on channel {}: {:?}", self.id, method);
                Error::InvalidConnectionState(ConnectionState::Closed)
            });
        self.internal_rpc.set_connection_closing();
        self.frames.drop_pending(error.clone());
        if let Some(resolver) = self.connection_status.connection_resolver() {
            resolver.swear(Err(error.clone()));
        }
        self.internal_rpc.send_connection_close_ok(error);
        Ok(())
    }

    fn on_connection_blocked_received(&self, _method: protocol::connection::Blocked) -> Result<()> {
        self.connection_status.block();
        Ok(())
    }

    fn on_connection_unblocked_received(
        &self,
        _method: protocol::connection::Unblocked,
    ) -> Result<()> {
        self.connection_status.unblock();
        self.wake();
        Ok(())
    }

    fn on_connection_close_ok_received(&self) -> Result<()> {
        self.internal_rpc
            .set_connection_closed(Error::InvalidConnectionState(ConnectionState::Closed));
        Ok(())
    }

    fn on_channel_open_ok_received(
        &self,
        _method: protocol::channel::OpenOk,
        resolver: PromiseResolver<Channel>,
        channel: Channel,
    ) -> Result<()> {
        self.set_state(ChannelState::Connected);
        resolver.swear(Ok(channel));
        Ok(())
    }

    fn on_channel_flow_received(&self, method: protocol::channel::Flow) -> Result<()> {
        self.status.set_send_flow(method.active);
        self.internal_rpc
            .register_internal_future(self.channel_flow_ok(ChannelFlowOkOptions {
                active: method.active,
            }))
    }

    fn on_channel_flow_ok_received(
        &self,
        method: protocol::channel::FlowOk,
        resolver: PromiseResolver<Boolean>,
    ) -> Result<()> {
        // Nothing to do here, the server just confirmed that we paused/resumed the receiving flow
        resolver.swear(Ok(method.active));
        Ok(())
    }

    fn on_channel_close_received(&self, method: protocol::channel::Close) -> Result<()> {
        let error = AMQPError::try_from(method.clone()).map(|error| {
            error!(
                "Channel closed on channel {} by {}:{} => {:?} => {}",
                self.id, method.class_id, method.method_id, error, method.reply_text
            );
            Error::ProtocolError(error)
        });
        if let Ok(error) = error.as_ref() {
            let _ = self.set_error(error.clone()); // Only error case here can be from spawning a thread in the default executor
        } else {
            self.set_closing();
        }
        let error = error.unwrap_or_else(|error| {<commit_msg>Fix regression in soft error handling<commit_after>use crate::{
    acknowledgement::Acknowledgements,
    auth::Credentials,
    basic_get_delivery::BasicGetDelivery,
    channel_closer::ChannelCloser,
    channel_receiver_state::DeliveryCause,
    channel_status::{ChannelState, ChannelStatus},
    connection_closer::ConnectionCloser,
    connection_status::{ConnectionState, ConnectionStep},
    consumer::Consumer,
    consumers::Consumers,
    executor::Executor,
    frames::{ExpectedReply, Frames},
    internal_rpc::InternalRPCHandle,
    message::{BasicGetMessage, BasicReturnMessage, Delivery},
    protocol::{self, AMQPClass, AMQPError, AMQPHardError},
    publisher_confirm::PublisherConfirm,
    queue::Queue,
    registry::Registry,
    returned_messages::ReturnedMessages,
    socket_state::SocketStateHandle,
    topology::RestoredChannel,
    topology_internal::ChannelDefinitionInternal,
    types::*,
    BasicProperties, Configuration, ConfirmationPromise, Connection, ConnectionStatus, DeliveryTag,
    Error, ExchangeKind, Promise, PromiseChain, PromiseResolver, Result,
};
use amq_protocol::frame::{AMQPContentHeader, AMQPFrame};
use log::{debug, error, info, log_enabled, trace, Level::Trace};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::{convert::TryFrom, fmt, sync::Arc};

#[derive(Clone)]
pub struct Channel {
    id: u16,
    configuration: Configuration,
    status: ChannelStatus,
    connection_status: ConnectionStatus,
    global_registry: Registry,
    local_registry: Registry,
    acknowledgements: Acknowledgements,
    consumers: Consumers,
    basic_get_delivery: BasicGetDelivery,
    returned_messages: ReturnedMessages,
    waker: SocketStateHandle,
    internal_rpc: InternalRPCHandle,
    frames: Frames,
    executor: Arc<dyn Executor>,
    channel_closer: Option<Arc<ChannelCloser>>,
    connection_closer: Option<Arc<ConnectionCloser>>,
}

impl PartialEq for Channel {
    fn eq(&self, other: &Self) -> bool {
        self.id == other.id
    }
}

impl fmt::Debug for Channel {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Channel")
            .field("id", &self.id)
            .field("configuration", &self.configuration)
            .field("status", &self.status)
            .field("connection_status", &self.connection_status)
            .field("acknowledgements", &self.acknowledgements)
            .field("consumers", &self.consumers)
            .field("basic_get_delivery", &self.basic_get_delivery)
            .field("returned_messages", &self.returned_messages)
            .field("frames", &self.frames)
            .field("executor", &self.executor)
            .finish()
    }
}

impl Channel {
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn new(
        channel_id: u16,
        configuration: Configuration,
        connection_status: ConnectionStatus,
        global_registry: Registry,
        waker: SocketStateHandle,
        internal_rpc: InternalRPCHandle,
        frames: Frames,
        executor: Arc<dyn Executor>,
        connection_closer: Option<Arc<ConnectionCloser>>,
    ) -> Channel {
        let returned_messages = ReturnedMessages::default();
        let status = ChannelStatus::default();
        let channel_closer = if channel_id == 0 {
            None
        } else {
            Some(Arc::new(ChannelCloser::new(
                channel_id,
                status.clone(),
                internal_rpc.clone(),
            )))
        };
        Channel {
            id: channel_id,
            configuration,
            status,
            connection_status,
            global_registry,
            local_registry: Registry::default(),
            acknowledgements: Acknowledgements::new(channel_id, returned_messages.clone()),
            consumers: Consumers::default(),
            basic_get_delivery: BasicGetDelivery::default(),
            returned_messages,
            waker,
            internal_rpc,
            frames,
            executor,
            channel_closer,
            connection_closer,
        }
    }

    pub fn status(&self) -> &ChannelStatus {
        &self.status
    }

    pub(crate) fn reset(&self) {
        // FIXME
    }

    pub(crate) async fn restore(
        &self,
        ch: &ChannelDefinitionInternal,
        c: &mut RestoredChannel,
    ) -> Result<()> {
        // First, redeclare all queues
        for queue in &ch.queues {
            if queue.is_declared() {
                c.queues.push(
                    self.queue_declare(
                        queue.name.as_str(),
                        queue.options.unwrap_or_default(),
                        queue.arguments.clone().unwrap_or_default(),
                    )
                    .await?,
                );
            }
        }

        // Second, redeclare all queues bindings
        for queue in &ch.queues {
            for binding in &queue.bindings {
                self.queue_bind(
                    queue.name.as_str(),
                    binding.source.as_str(),
                    binding.routing_key.as_str(),
                    QueueBindOptions::default(),
                    binding.arguments.clone(),
                )
                .await?;
            }
        }

        // Third, redeclare all consumers
        for consumer in &ch.consumers {
            let original = consumer.original();
            if let Some(original) = original.as_ref() {
                original.reset();
            }
            c.consumers.push(
                self.do_basic_consume(
                    consumer.queue.as_str(),
                    consumer.tag.as_str(),
                    consumer.options,
                    consumer.arguments.clone(),
                    original,
                )
                .await?,
            );
        }

        // Fourth, reemit pending basic_get
        if let Some(original) = self.basic_get_delivery.recover() {
            self.do_basic_get(
                original.queue.as_str(),
                original.options,
                Some(original.resolver),
            )
            .await?;
        }

        Ok(())
    }

    fn set_closing(&self, error: Option<Error>) {
        self.set_state(ChannelState::Closing);
        if let Some(error) = error {
            self.error_publisher_confirms(error.clone());
            let _ = self.error_consumers(error.clone()); // ignore the error here, only happens with default executor if we cannot spawn a thread
        } else {
            self.consumers.start_cancel();
        }
    }

    fn set_closed(&self, error: Error) -> Result<()> {
        self.set_state(ChannelState::Closed);
        self.error_publisher_confirms(error.clone());
        let res = self.cancel_consumers();
        self.internal_rpc.remove_channel(self.id, error);
        res
    }

    fn set_error(&self, error: Error) -> Result<()> {
        self.set_state(ChannelState::Error);
        self.error_publisher_confirms(error.clone());
        let res = self.error_consumers(error.clone());
        self.internal_rpc.remove_channel(self.id, error);
        res
    }

    pub(crate) fn error_publisher_confirms(&self, error: Error) {
        self.acknowledgements.on_channel_error(error);
    }

    pub(crate) fn cancel_consumers(&self) -> Result<()> {
        self.consumers.cancel()
    }

    pub(crate) fn error_consumers(&self, error: Error) -> Result<()> {
        self.consumers.error(error)
    }

    pub(crate) fn set_state(&self, state: ChannelState) {
        self.status.set_state(state);
    }

    pub fn id(&self) -> u16 {
        self.id
    }

    pub(crate) fn clone_internal(&self) -> Self {
        Self {
            id: self.id,
            configuration: self.configuration.clone(),
            status: self.status.clone(),
            connection_status: self.connection_status.clone(),
            global_registry: self.global_registry.clone(),
            local_registry: self.local_registry.clone(),
            acknowledgements: self.acknowledgements.clone(),
            consumers: self.consumers.clone(),
            basic_get_delivery: self.basic_get_delivery.clone(),
            returned_messages: self.returned_messages.clone(),
            waker: self.waker.clone(),
            internal_rpc: self.internal_rpc.clone(),
            frames: self.frames.clone(),
            executor: self.executor.clone(),
            channel_closer: None,
            connection_closer: self.connection_closer.clone(),
        }
    }

    fn wake(&self) {
        trace!("channel {} wake", self.id);
        self.waker.wake()
    }

    fn assert_channel0(&self, class_id: u16, method_id: u16) -> Result<()> {
        if self.id == 0 {
            Ok(())
        } else {
            error!(
                "Got a connection frame on channel {}, closing connection",
                self.id
            );
            let error = AMQPError::new(
                AMQPHardError::COMMANDINVALID.into(),
                format!("connection frame received on channel {}", self.id).into(),
            );
            self.internal_rpc.close_connection(
                error.get_id(),
                error.get_message().to_string(),
                class_id,
                method_id,
            );
            Err(Error::ProtocolError(error))
        }
    }

    pub fn close(&self, reply_code: ShortUInt, reply_text: &str) -> Promise<()> {
        self.do_channel_close(reply_code, reply_text, 0, 0)
    }

    pub fn basic_consume(
        &self,
        queue: &str,
        consumer_tag: &str,
        options: BasicConsumeOptions,
        arguments: FieldTable,
    ) -> PromiseChain<Consumer> {
        self.do_basic_consume(queue, consumer_tag, options, arguments, None)
    }

    pub fn basic_get(
        &self,
        queue: &str,
        options: BasicGetOptions,
    ) -> PromiseChain<Option<BasicGetMessage>> {
        self.do_basic_get(queue, options, None)
    }

    pub fn exchange_declare(
        &self,
        exchange: &str,
        kind: ExchangeKind,
        options: ExchangeDeclareOptions,
        arguments: FieldTable,
    ) -> Promise<()> {
        self.do_exchange_declare(exchange, kind.kind(), options, arguments, kind.clone())
    }

    pub fn wait_for_confirms(&self) -> ConfirmationPromise<Vec<BasicReturnMessage>> {
        if let Some(promise) = self.acknowledgements.get_last_pending() {
            trace!("Waiting for pending confirms");
            let returned_messages = self.returned_messages.clone();
            promise.traverse(move |_| Ok(returned_messages.drain()))
        } else {
            trace!("No confirms to wait for");
            ConfirmationPromise::new_with_data(Ok(Vec::default()))
        }
    }

    #[cfg(test)]
    pub(crate) fn register_queue(
        &self,
        name: ShortString,
        options: QueueDeclareOptions,
        arguments: FieldTable,
    ) {
        self.local_registry.register_queue(name, options, arguments);
    }

    #[cfg(test)]
    pub(crate) fn register_consumer(&self, tag: ShortString, consumer: Consumer) {
        self.consumers.register(tag, consumer);
    }

    pub(crate) fn send_method_frame(
        &self,
        method: AMQPClass,
        resolver: PromiseResolver<()>,
        expected_reply: Option<ExpectedReply>,
    ) {
        self.send_frame(AMQPFrame::Method(self.id, method), resolver, expected_reply);
    }

    pub(crate) fn send_frame(
        &self,
        frame: AMQPFrame,
        resolver: PromiseResolver<()>,
        expected_reply: Option<ExpectedReply>,
    ) {
        trace!("channel {} send_frame", self.id);
        self.frames.push(self.id, frame, resolver, expected_reply);
        self.wake();
    }

    fn send_method_frame_with_body(
        &self,
        method: AMQPClass,
        payload: Vec<u8>,
        properties: BasicProperties,
        publisher_confirms_result: Option<PublisherConfirm>,
    ) -> Result<PromiseChain<PublisherConfirm>> {
        let class_id = method.get_amqp_class_id();
        let header = AMQPContentHeader {
            class_id,
            weight: 0,
            body_size: payload.len() as u64,
            properties,
        };
        let frame_max = self.configuration.frame_max();
        let mut frames = vec![
            AMQPFrame::Method(self.id, method),
            AMQPFrame::Header(self.id, class_id, Box::new(header)),
        ];

        // a content body frame 8 bytes of overhead
        frames.extend(
            payload
                .as_slice()
                .chunks(frame_max as usize - 8)
                .map(|chunk| AMQPFrame::Body(self.id, chunk.into())),
        );

        // tweak to make rustc happy
        let data = Arc::new(Mutex::new((
            publisher_confirms_result,
            self.returned_messages.clone(),
        )));

        trace!("channel {} send_frames", self.id);
        let promise = self.frames.push_frames(frames);
        self.wake();
        Ok(promise.traverse(move |res| {
            res.map(|()| {
                let mut data = data.lock();
                data.0
                    .take()
                    .unwrap_or_else(|| PublisherConfirm::not_requested(data.1.clone()))
            })
        }))
    }

    fn handle_invalid_contents(&self, error: String, class_id: u16, method_id: u16) -> Result<()> {
        error!("{}", error);
        let error = AMQPError::new(AMQPHardError::UNEXPECTEDFRAME.into(), error.into());
        self.internal_rpc.close_connection(
            error.get_id(),
            error.get_message().to_string(),
            class_id,
            method_id,
        );
        Err(Error::ProtocolError(error))
    }

    pub(crate) fn handle_content_header_frame(
        &self,
        class_id: u16,
        size: LongLongUInt,
        properties: BasicProperties,
    ) -> Result<()> {
        self.status.set_content_length(
            self.id,
            class_id,
            size,
            |delivery_cause, confirm_mode| {
                match delivery_cause {
                    DeliveryCause::Consume(consumer_tag) => {
                        self.consumers.handle_content_header_frame(
                            &self,
                            consumer_tag,
                            size,
                            properties,
                        )?;
                    }
                    DeliveryCause::Get => {
                        self.basic_get_delivery
                            .handle_content_header_frame(size, properties);
                    }
                    DeliveryCause::Return => {
                        self.returned_messages.handle_content_header_frame(
                            size,
                            properties,
                            confirm_mode,
                        );
                    }
                }
                Ok(())
            },
            |msg| {
                error!("{}", msg);
                let error = AMQPError::new(AMQPHardError::FRAMEERROR.into(), msg.into());
                self.internal_rpc.close_connection(
                    error.get_id(),
                    error.get_message().to_string(),
                    class_id,
                    0,
                );
                self.set_error(Error::ProtocolError(error))
            },
            |msg| self.handle_invalid_contents(msg, class_id, 0),
        )
    }

    pub(crate) fn handle_body_frame(&self, payload: Vec<u8>) -> Result<()> {
        self.status.receive(
            self.id,
            payload.len() as LongLongUInt,
            |delivery_cause, remaining_size, confirm_mode| {
                match delivery_cause {
                    DeliveryCause::Consume(consumer_tag) => {
                        self.consumers.handle_body_frame(
                            &self,
                            consumer_tag,
                            remaining_size,
                            payload,
                        )?;
                    }
                    DeliveryCause::Get => {
                        self.basic_get_delivery
                            .handle_body_frame(remaining_size, payload);
                    }
                    DeliveryCause::Return => {
                        self.returned_messages.handle_body_frame(
                            remaining_size,
                            payload,
                            confirm_mode,
                        );
                    }
                }
                Ok(())
            },
            |msg| self.handle_invalid_contents(msg, 0, 0),
        )
    }

    pub(crate) fn topology(&self) -> ChannelDefinitionInternal {
        ChannelDefinitionInternal {
            channel: Some(self.clone()),
            queues: self.local_registry.queues_topology(true),
            consumers: self.consumers.topology(),
        }
    }

    fn before_basic_publish(&self) -> Option<PublisherConfirm> {
        if self.status.confirm() {
            Some(self.acknowledgements.register_pending())
        } else {
            None
        }
    }

    fn before_basic_cancel(&self, consumer_tag: &str) {
        self.consumers.start_cancel_one(consumer_tag);
    }

    fn acknowledgement_error(&self, error: AMQPError, class_id: u16, method_id: u16) -> Result<()> {
        error!("Got a bad acknowledgement from server, closing channel");
        self.internal_rpc
            .register_internal_future(self.do_channel_close(
                error.get_id(),
                error.get_message().as_str(),
                class_id,
                method_id,
            ))?;
        Err(Error::ProtocolError(error))
    }

    fn on_connection_start_ok_sent(
        &self,
        resolver: PromiseResolver<Connection>,
        connection: Connection,
        credentials: Credentials,
    ) -> Result<()> {
        self.connection_status
            .set_connection_step(ConnectionStep::StartOk(resolver, connection, credentials));
        Ok(())
    }

    fn on_connection_open_sent(&self, resolver: PromiseResolver<Connection>) -> Result<()> {
        self.connection_status
            .set_connection_step(ConnectionStep::Open(resolver));
        Ok(())
    }

    fn on_connection_close_ok_sent(&self, error: Error) -> Result<()> {
        if let Error::ProtocolError(_) = error {
            self.internal_rpc.set_connection_error(error);
        } else {
            self.internal_rpc.set_connection_closed(error);
        }
        Ok(())
    }

    fn next_expected_close_ok_reply(&self) -> Option<Reply> {
        self.frames
            .next_expected_close_ok_reply(self.id, Error::InvalidChannelState(ChannelState::Closed))
    }

    fn before_channel_close(&self) {
        self.set_closing(None);
    }

    fn on_channel_close_ok_sent(&self, error: Error) -> Result<()> {
        self.set_closed(error)
    }

    fn on_basic_recover_async_sent(&self) -> Result<()> {
        self.consumers.drop_prefetched_messages()
    }

    fn on_basic_ack_sent(&self, multiple: bool, delivery_tag: DeliveryTag) -> Result<()> {
        if multiple && delivery_tag == 0 {
            self.consumers.drop_prefetched_messages()
        } else {
            Ok(())
        }
    }

    fn on_basic_nack_sent(&self, multiple: bool, delivery_tag: DeliveryTag) -> Result<()> {
        if multiple && delivery_tag == 0 {
            self.consumers.drop_prefetched_messages()
        } else {
            Ok(())
        }
    }

    fn tune_connection_configuration(&self, channel_max: u16, frame_max: u32, heartbeat: u16) {
        // If we disable the heartbeat (0) but the server don't, follow it and enable it too
        // If both us and the server want heartbeat enabled, pick the lowest value.
        if self.configuration.heartbeat() == 0
            || (heartbeat != 0 && heartbeat < self.configuration.heartbeat())
        {
            self.configuration.set_heartbeat(heartbeat);
        }

        if channel_max != 0 {
            // 0 means we want to take the server's value
            // If both us and the server specified a channel_max, pick the lowest value.
            if self.configuration.channel_max() == 0
                || channel_max < self.configuration.channel_max()
            {
                self.configuration.set_channel_max(channel_max);
            }
        }
        if self.configuration.channel_max() == 0 {
            self.configuration.set_channel_max(u16::max_value());
        }

        if frame_max != 0 {
            // 0 means we want to take the server's value
            // If both us and the server specified a frame_max, pick the lowest value.
            if self.configuration.frame_max() == 0 || frame_max < self.configuration.frame_max() {
                self.configuration.set_frame_max(frame_max);
            }
        }
        if self.configuration.frame_max() == 0 {
            self.configuration.set_frame_max(u32::max_value());
        }
    }

    fn on_connection_start_received(&self, method: protocol::connection::Start) -> Result<()> {
        trace!("Server sent connection::Start: {:?}", method);
        let state = self.connection_status.state();
        if let (
            ConnectionState::Connecting,
            Some(ConnectionStep::ProtocolHeader(
                resolver,
                connection,
                credentials,
                mechanism,
                mut options,
            )),
        ) = (state.clone(), self.connection_status.connection_step())
        {
            let mechanism_str = mechanism.to_string();
            let locale = options.locale.clone();

            if !method
                .mechanisms
                .split_whitespace()
                .any(|m| m == mechanism_str)
            {
                error!("unsupported mechanism: {}", mechanism);
            }
            if !method.locales.split_whitespace().any(|l| l == locale) {
                error!("unsupported locale: {}", mechanism);
            }

            if !options.client_properties.contains_key("product")
                || !options.client_properties.contains_key("version")
            {
                options.client_properties.insert(
                    "product".into(),
                    AMQPValue::LongString(env!("CARGO_PKG_NAME").into()),
                );
                options.client_properties.insert(
                    "version".into(),
                    AMQPValue::LongString(env!("CARGO_PKG_VERSION").into()),
                );
            }

            options
                .client_properties
                .insert("platform".into(), AMQPValue::LongString("rust".into()));

            let mut capabilities = FieldTable::default();
            capabilities.insert("publisher_confirms".into(), true.into());
            capabilities.insert("exchange_exchange_bindings".into(), true.into());
            capabilities.insert("basic.nack".into(), true.into());
            capabilities.insert("consumer_cancel_notify".into(), true.into());
            capabilities.insert("connection.blocked".into(), true.into());
            capabilities.insert("consumer_priorities".into(), true.into());
            capabilities.insert("authentication_failure_close".into(), true.into());
            capabilities.insert("per_consumer_qos".into(), true.into());
            capabilities.insert("direct_reply_to".into(), true.into());

            options
                .client_properties
                .insert("capabilities".into(), AMQPValue::FieldTable(capabilities));

            self.internal_rpc
                .register_internal_future(self.connection_start_ok(
                    options.client_properties,
                    &mechanism_str,
                    &credentials.sasl_auth_string(mechanism),
                    &locale,
                    resolver,
                    connection,
                    credentials,
                ))
        } else {
            error!("Invalid state: {:?}", state);
            let error = Error::InvalidConnectionState(state);
            self.internal_rpc.set_connection_error(error.clone());
            Err(error)
        }
    }

    fn on_connection_secure_received(&self, method: protocol::connection::Secure) -> Result<()> {
        trace!("Server sent connection::Secure: {:?}", method);

        let state = self.connection_status.state();
        if let (ConnectionState::Connecting, Some(ConnectionStep::StartOk(.., credentials))) =
            (state.clone(), self.connection_status.connection_step())
        {
            self.internal_rpc.register_internal_future(
                self.connection_secure_ok(&credentials.rabbit_cr_demo_answer()),
            )
        } else {
            error!("Invalid state: {:?}", state);
            let error = Error::InvalidConnectionState(state);
            self.internal_rpc.set_connection_error(error.clone());
            Err(error)
        }
    }

    fn on_connection_tune_received(&self, method: protocol::connection::Tune) -> Result<()> {
        debug!("Server sent Connection::Tune: {:?}", method);

        let state = self.connection_status.state();
        if let (
            ConnectionState::Connecting,
            Some(ConnectionStep::StartOk(resolver, connection, _)),
        ) = (state.clone(), self.connection_status.connection_step())
        {
            self.tune_connection_configuration(
                method.channel_max,
                method.frame_max,
                method.heartbeat,
            );

            self.internal_rpc
                .register_internal_future(self.connection_tune_ok(
                    self.configuration.channel_max(),
                    self.configuration.frame_max(),
                    self.configuration.heartbeat(),
                ))?;
            self.internal_rpc
                .register_internal_future(self.connection_open(
                    &self.connection_status.vhost(),
                    connection,
                    resolver,
                ))
        } else {
            error!("Invalid state: {:?}", state);
            let error = Error::InvalidConnectionState(state);
            self.internal_rpc.set_connection_error(error.clone());
            Err(error)
        }
    }

    fn on_connection_open_ok_received(
        &self,
        _: protocol::connection::OpenOk,
        connection: Connection,
    ) -> Result<()> {
        let state = self.connection_status.state();
        if let (ConnectionState::Connecting, Some(ConnectionStep::Open(resolver))) =
            (state.clone(), self.connection_status.connection_step())
        {
            self.connection_status.set_state(ConnectionState::Connected);
            resolver.swear(Ok(connection));
            Ok(())
        } else {
            error!("Invalid state: {:?}", state);
            let error = Error::InvalidConnectionState(state);
            self.internal_rpc.set_connection_error(error.clone());
            Err(error)
        }
    }

    fn on_connection_close_received(&self, method: protocol::connection::Close) -> Result<()> {
        let error = AMQPError::try_from(method.clone())
            .map(|error| {
                error!(
                    "Connection closed on channel {} by {}:{} => {:?} => {}",
                    self.id, method.class_id, method.method_id, error, method.reply_text
                );
                Error::ProtocolError(error)
            })
            .unwrap_or_else(|error| {
                error!("{}", error);
                info!("Connection closed on channel {}: {:?}", self.id, method);
                Error::InvalidConnectionState(ConnectionState::Closed)
            });
        self.internal_rpc.set_connection_closing();
        self.frames.drop_pending(error.clone());
        if let Some(resolver) = self.connection_status.connection_resolver() {
            resolver.swear(Err(error.clone()));
        }
        self.internal_rpc.send_connection_close_ok(error);
        Ok(())
    }

    fn on_connection_blocked_received(&self, _method: protocol::connection::Blocked) -> Result<()> {
        self.connection_status.block();
        Ok(())
    }

    fn on_connection_unblocked_received(
        &self,
        _method: protocol::connection::Unblocked,
    ) -> Result<()> {
        self.connection_status.unblock();
        self.wake();
        Ok(())
    }

    fn on_connection_close_ok_received(&self) -> Result<()> {
        self.internal_rpc
            .set_connection_closed(Error::InvalidConnectionState(ConnectionState::Closed));
        Ok(())
    }

    fn on_channel_open_ok_received(
        &self,
        _method: protocol::channel::OpenOk,
        resolver: PromiseResolver<Channel>,
        channel: Channel,
    ) -> Result<()> {
        self.set_state(ChannelState::Connected);
        resolver.swear(Ok(channel));
        Ok(())
    }

    fn on_channel_flow_received(&self, method: protocol::channel::Flow) -> Result<()> {
        self.status.set_send_flow(method.active);
        self.internal_rpc
            .register_internal_future(self.channel_flow_ok(ChannelFlowOkOptions {
                active: method.active,
            }))
    }

    fn on_channel_flow_ok_received(
        &self,
        method: protocol::channel::FlowOk,
        resolver: PromiseResolver<Boolean>,
    ) -> Result<()> {
        // Nothing to do here, the server just confirmed that we paused/resumed the receiving flow
        resolver.swear(Ok(method.active));
        Ok(())
    }

    fn on_channel_close_received(&self, method: protocol::channel::Close) -> Result<()> {
        let error = AMQPError::try_from(method.clone()).map(|error| {
            error!(
                "Channel closed on channel {} by {}:{} => {:?} => {}",
                self.id, method.class_id, method.method_id, error, method.reply_text
            );
            Error::ProtocolError(error)
        });
        self.set_closing(error.clone().ok());
        let error = error.unwrap_or_else(|error| {
{'fraction_non_alphanumeric': 0.11583699853391977, 'fraction_numerical': 0.0020325203252032522, 'mean_word_length': 1.847080586254328, 'pattern_counts': {'":': 0, '<': 130, '<?xml version=': 0, '>': 232, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1946967', 'n_tokens_mistral': 16553, 'n_tokens_neox': 15942, 'n_words': 2772}