<commit_before>pub use lapin_async::channel::BasicProperties;
pub use lapin_async::channel::options::*;

use futures::{Async, Future, future, Poll, Stream, task};
use lapin_async;
use lapin_async::channel::Channel as InnerChannel;
use lapin_async::channel_status::ChannelState;
use lapin_async::queue::QueueStats;
use lapin_async::requests::RequestId;
use log::{debug, trace};
use parking_lot::Mutex;
use tokio_io::{AsyncRead, AsyncWrite};

use std::sync::Arc;

use crate::consumer::Consumer;
use crate::error::{Error, ErrorKind};
use crate::message::BasicGetMessage;
use crate::queue::Queue;
use crate::transport::*;
use crate::types::*;

pub type RequestResult = Result<Option<RequestId>, lapin_async::error::Error>;

/// `Channel` provides methods to act on a channel, such as managing queues
//#[derive(Clone)]
pub struct Channel<T> {
  pub transport: Arc<Mutex<AMQPTransport<T>>>,
      inner:     InnerChannel,
}

impl<T> Clone for Channel<T>
where T: Send {
  fn clone(&self) -> Channel<T> {
    Channel {
      transport: self.transport.clone(),
      inner:     self.inner.clone(),
    }
  }
}

impl<T: AsyncRead+AsyncWrite+Send+Sync+'static> Channel<T> {
    /// create a channel
    pub fn create(inner: Result<InnerChannel, lapin_async::error::Error>, transport: Arc<Mutex<AMQPTransport<T>>>) -> impl Future<Item = Self, Error = Error> + Send + 'static {
        future::result(inner.map(|inner| Channel { transport, inner }).map_err(|err| ErrorKind::ProtocolError("Failed to create channel".to_string(), err).into())).and_then(|channel| {
            let request_id = channel.inner.channel_open();
            let inner = channel.inner.clone();
            channel.run_on_locked_transport("create", "Could not create channel", request_id).and_then(move |_| {
                future::poll_fn(move || {
                    match inner.status.state() {
                        ChannelState::Connected => Ok(Async::Ready(())),
                        ChannelState::Error | ChannelState::Closed => {
                            Err(ErrorKind::ChannelOpenFailed.into())
                        },
                        _ => {
                            task::current().notify();
                            Ok(Async::NotReady)
                        }
                    }
                })
            }).map(move |_| {
                channel
            })
        })
    }

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

    /// request access
    ///
    /// returns a future that resolves once the access is granted
    pub fn access_request(&self, realm: &str, options: AccessRequestOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.access_request(realm, options);

        self.run_on_locked_transport("access_request", "Could not request access", request_id).map(|_| ())
    }

    /// declares an exchange
    ///
    /// returns a future that resolves once the exchange is available
    pub fn exchange_declare(&self, name: &str, exchange_type: &str, options: ExchangeDeclareOptions, arguments: FieldTable) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.exchange_declare(name, exchange_type, options, arguments);

        self.run_on_locked_transport("exchange_declare", "Could not declare exchange", request_id).map(|_| ())
    }

    /// deletes an exchange
    ///
    /// returns a future that resolves once the exchange is deleted
    pub fn exchange_delete(&self, name: &str, options: ExchangeDeleteOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.exchange_delete(name, options);

        self.run_on_locked_transport("exchange_delete", "Could not delete exchange", request_id).map(|_| ())
    }

    /// binds an exchange to another exchange
    ///
    /// returns a future that resolves once the exchanges are bound
    pub fn exchange_bind(&self, destination: &str, source: &str, routing_key: &str, options: ExchangeBindOptions, arguments: FieldTable) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.exchange_bind(destination, source, routing_key, options, arguments);

        self.run_on_locked_transport("exchange_bind", "Could not bind exchange", request_id).map(|_| ())
    }

    /// unbinds an exchange from another one
    ///
    /// returns a future that resolves once the exchanges are unbound
    pub fn exchange_unbind(&self, destination: &str, source: &str, routing_key: &str, options: ExchangeUnbindOptions, arguments: FieldTable) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.exchange_unbind(destination, source, routing_key, options, arguments);

        self.run_on_locked_transport("exchange_unbind", "Could not unbind exchange", request_id).map(|_| ())
    }

    /// declares a queue
    ///
    /// returns a future that resolves once the queue is available
    ///
    /// the `mandatory` and `ìmmediate` options can be set to true,
    /// but the return message will not be handled
    pub fn queue_declare(&self, name: &str, options: QueueDeclareOptions, arguments: FieldTable) -> impl Future<Item = Queue, Error = Error> + Send + 'static {
        let request_id = self.inner.queue_declare(name, options, arguments);
        let inner = self.inner.clone();

        self.run_on_locked_transport("queue_declare", "Could not declare queue", request_id).and_then(move |request_id| {
            future::poll_fn(move || {
              if let Some(queue) = inner.generated_names.get(request_id.expect("expected request_id")) {
                let QueueStats {consumer_count, message_count} = inner.queues.get_stats(&queue);
                return Ok(Async::Ready(Queue::new(queue, consumer_count, message_count)))
              } else {
                task::current().notify();
                return Ok(Async::NotReady)
              }
            })
        })
    }

    /// binds a queue to an exchange
    ///
    /// returns a future that resolves once the queue is bound to the exchange
    pub fn queue_bind(&self, name: &str, exchange: &str, routing_key: &str, options: QueueBindOptions, arguments: FieldTable) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.queue_bind(name, exchange, routing_key, options, arguments);

        self.run_on_locked_transport("queue_bind", "Could not bind queue", request_id).map(|_| ())
    }

    /// unbinds a queue from the exchange
    ///
    /// returns a future that resolves once the queue is unbound from the exchange
    pub fn queue_unbind(&self, name: &str, exchange: &str, routing_key: &str, arguments: FieldTable) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.queue_unbind(name, exchange, routing_key, arguments);

        self.run_on_locked_transport("queue_unbind", "Could not unbind queue from the exchange", request_id).map(|_| ())
    }

    /// sets up confirm extension for this channel
    pub fn confirm_select(&self, options: ConfirmSelectOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.confirm_select(options);

        self.run_on_locked_transport("confirm_select", "Could not activate confirm extension", request_id).map(|_| ())
    }

    /// specifies quality of service for a channel
    pub fn basic_qos(&self, prefetch_count: ShortUInt, options: BasicQosOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_qos(prefetch_count, options);

        self.run_on_locked_transport("basic_qos", "Could not setup qos", request_id).map(|_| ())
    }

    /// publishes a message on a queue
    ///
    /// the future's result is:
    /// - `Some(request_id)` if we're on a confirm channel and the message was ack'd
    /// - `None` if we're not on a confirm channel or the message was nack'd
    pub fn basic_publish(&self, exchange: &str, routing_key: &str, payload: Vec<u8>, options: BasicPublishOptions, properties: BasicProperties) -> impl Future<Item = Option<RequestId>, Error = Error> + Send + 'static {
        let request_id = self.inner.basic_publish(exchange, routing_key, options, payload, properties);

        self.run_on_locked_transport_full("basic_publish", "Could not publish", request_id, move |channel, delivery_tag| {
          if channel.status.confirm() {
            if channel.acknowledgements.is_acked(delivery_tag) {
              Ok(Async::Ready(Some(delivery_tag)))
            } else if channel.acknowledgements.is_nacked(delivery_tag) {
              Ok(Async::Ready(None))
            } else {
              debug!("message with tag {} still in unacked", delivery_tag);
              task::current().notify();
              Ok(Async::NotReady)
            }
          } else {
            Ok(Async::Ready(None))
          }
        })
    }

    /// creates a consumer stream
    ///
    /// returns a future of a `Consumer` that resolves once the method succeeds
    ///
    /// `Consumer` implements `futures::Stream`, so it can be used with any of
    /// the usual combinators
    pub fn basic_consume(&self, queue: &Queue, consumer_tag: &str, options: BasicConsumeOptions, arguments: FieldTable) -> impl Future<Item = Consumer<T>, Error = Error> + Send + 'static {
        let consumer_tag = consumer_tag.to_string();
        let queue_name = queue.name();
        let mut consumer = Consumer::new(self.transport.clone(), self.id(), queue.name(), consumer_tag.to_owned());
        let subscriber = consumer.subscriber();
        let request_id = self.inner.basic_consume(&queue_name, &consumer_tag, options, arguments, Box::new(subscriber));
        let inner = self.inner.clone();

        self.run_on_locked_transport("basic_consume", "Could not start consumer", request_id).and_then(move |request_id| {
            future::poll_fn(move || {
              if let Some(consumer_tag) = inner.generated_names.get(request_id.expect("expected request_id")) {
                return Ok(Async::Ready(consumer_tag))
              } else {
                task::current().notify();
                return Ok(Async::NotReady)
              }
            })
          }).map(|consumer_tag| {
            trace!("basic_consume received response, returning consumer");
            consumer.update_consumer_tag(consumer_tag);
            consumer
        })
    }

    /// acks a message
    pub fn basic_ack(&self, delivery_tag: u64, multiple: bool) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_ack(delivery_tag, BasicAckOptions { multiple });

        self.run_on_locked_transport("basic_ack", "Could not ack message", request_id).map(|_| ())
    }

    /// nacks a message
    pub fn basic_nack(&self, delivery_tag: u64, multiple: bool, requeue: bool) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_nack(delivery_tag, BasicNackOptions { multiple, requeue });

        self.run_on_locked_transport("basic_nack", "Could not nack message", request_id).map(|_| ())
    }

    /// rejects a message
    pub fn basic_reject(&self, delivery_tag: u64, options: BasicRejectOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_reject(delivery_tag, options);

        self.run_on_locked_transport("basic_reject", "Could not reject message", request_id).map(|_| ())
    }<commit_msg>futures: implement missing basic methods<commit_after>pub use lapin_async::channel::BasicProperties;
pub use lapin_async::channel::options::*;

use futures::{Async, Future, future, Poll, Stream, task};
use lapin_async;
use lapin_async::channel::Channel as InnerChannel;
use lapin_async::channel_status::ChannelState;
use lapin_async::queue::QueueStats;
use lapin_async::requests::RequestId;
use log::{debug, trace};
use parking_lot::Mutex;
use tokio_io::{AsyncRead, AsyncWrite};

use std::sync::Arc;

use crate::consumer::Consumer;
use crate::error::{Error, ErrorKind};
use crate::message::BasicGetMessage;
use crate::queue::Queue;
use crate::transport::*;
use crate::types::*;

pub type RequestResult = Result<Option<RequestId>, lapin_async::error::Error>;

/// `Channel` provides methods to act on a channel, such as managing queues
//#[derive(Clone)]
pub struct Channel<T> {
  pub transport: Arc<Mutex<AMQPTransport<T>>>,
      inner:     InnerChannel,
}

impl<T> Clone for Channel<T>
where T: Send {
  fn clone(&self) -> Channel<T> {
    Channel {
      transport: self.transport.clone(),
      inner:     self.inner.clone(),
    }
  }
}

impl<T: AsyncRead+AsyncWrite+Send+Sync+'static> Channel<T> {
    /// create a channel
    pub fn create(inner: Result<InnerChannel, lapin_async::error::Error>, transport: Arc<Mutex<AMQPTransport<T>>>) -> impl Future<Item = Self, Error = Error> + Send + 'static {
        future::result(inner.map(|inner| Channel { transport, inner }).map_err(|err| ErrorKind::ProtocolError("Failed to create channel".to_string(), err).into())).and_then(|channel| {
            let request_id = channel.inner.channel_open();
            let inner = channel.inner.clone();
            channel.run_on_locked_transport("create", "Could not create channel", request_id).and_then(move |_| {
                future::poll_fn(move || {
                    match inner.status.state() {
                        ChannelState::Connected => Ok(Async::Ready(())),
                        ChannelState::Error | ChannelState::Closed => {
                            Err(ErrorKind::ChannelOpenFailed.into())
                        },
                        _ => {
                            task::current().notify();
                            Ok(Async::NotReady)
                        }
                    }
                })
            }).map(move |_| {
                channel
            })
        })
    }

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

    /// request access
    ///
    /// returns a future that resolves once the access is granted
    pub fn access_request(&self, realm: &str, options: AccessRequestOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.access_request(realm, options);

        self.run_on_locked_transport("access_request", "Could not request access", request_id).map(|_| ())
    }

    /// declares an exchange
    ///
    /// returns a future that resolves once the exchange is available
    pub fn exchange_declare(&self, name: &str, exchange_type: &str, options: ExchangeDeclareOptions, arguments: FieldTable) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.exchange_declare(name, exchange_type, options, arguments);

        self.run_on_locked_transport("exchange_declare", "Could not declare exchange", request_id).map(|_| ())
    }

    /// deletes an exchange
    ///
    /// returns a future that resolves once the exchange is deleted
    pub fn exchange_delete(&self, name: &str, options: ExchangeDeleteOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.exchange_delete(name, options);

        self.run_on_locked_transport("exchange_delete", "Could not delete exchange", request_id).map(|_| ())
    }

    /// binds an exchange to another exchange
    ///
    /// returns a future that resolves once the exchanges are bound
    pub fn exchange_bind(&self, destination: &str, source: &str, routing_key: &str, options: ExchangeBindOptions, arguments: FieldTable) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.exchange_bind(destination, source, routing_key, options, arguments);

        self.run_on_locked_transport("exchange_bind", "Could not bind exchange", request_id).map(|_| ())
    }

    /// unbinds an exchange from another one
    ///
    /// returns a future that resolves once the exchanges are unbound
    pub fn exchange_unbind(&self, destination: &str, source: &str, routing_key: &str, options: ExchangeUnbindOptions, arguments: FieldTable) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.exchange_unbind(destination, source, routing_key, options, arguments);

        self.run_on_locked_transport("exchange_unbind", "Could not unbind exchange", request_id).map(|_| ())
    }

    /// declares a queue
    ///
    /// returns a future that resolves once the queue is available
    ///
    /// the `mandatory` and `ìmmediate` options can be set to true,
    /// but the return message will not be handled
    pub fn queue_declare(&self, name: &str, options: QueueDeclareOptions, arguments: FieldTable) -> impl Future<Item = Queue, Error = Error> + Send + 'static {
        let request_id = self.inner.queue_declare(name, options, arguments);
        let inner = self.inner.clone();

        self.run_on_locked_transport("queue_declare", "Could not declare queue", request_id).and_then(move |request_id| {
            future::poll_fn(move || {
              if let Some(queue) = inner.generated_names.get(request_id.expect("expected request_id")) {
                let QueueStats {consumer_count, message_count} = inner.queues.get_stats(&queue);
                return Ok(Async::Ready(Queue::new(queue, consumer_count, message_count)))
              } else {
                task::current().notify();
                return Ok(Async::NotReady)
              }
            })
        })
    }

    /// binds a queue to an exchange
    ///
    /// returns a future that resolves once the queue is bound to the exchange
    pub fn queue_bind(&self, name: &str, exchange: &str, routing_key: &str, options: QueueBindOptions, arguments: FieldTable) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.queue_bind(name, exchange, routing_key, options, arguments);

        self.run_on_locked_transport("queue_bind", "Could not bind queue", request_id).map(|_| ())
    }

    /// unbinds a queue from the exchange
    ///
    /// returns a future that resolves once the queue is unbound from the exchange
    pub fn queue_unbind(&self, name: &str, exchange: &str, routing_key: &str, arguments: FieldTable) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.queue_unbind(name, exchange, routing_key, arguments);

        self.run_on_locked_transport("queue_unbind", "Could not unbind queue from the exchange", request_id).map(|_| ())
    }

    /// sets up confirm extension for this channel
    pub fn confirm_select(&self, options: ConfirmSelectOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.confirm_select(options);

        self.run_on_locked_transport("confirm_select", "Could not activate confirm extension", request_id).map(|_| ())
    }

    /// specifies quality of service for a channel
    pub fn basic_qos(&self, prefetch_count: ShortUInt, options: BasicQosOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_qos(prefetch_count, options);

        self.run_on_locked_transport("basic_qos", "Could not setup qos", request_id).map(|_| ())
    }

    /// publishes a message on a queue
    ///
    /// the future's result is:
    /// - `Some(request_id)` if we're on a confirm channel and the message was ack'd
    /// - `None` if we're not on a confirm channel or the message was nack'd
    pub fn basic_publish(&self, exchange: &str, routing_key: &str, payload: Vec<u8>, options: BasicPublishOptions, properties: BasicProperties) -> impl Future<Item = Option<RequestId>, Error = Error> + Send + 'static {
        let request_id = self.inner.basic_publish(exchange, routing_key, options, payload, properties);

        self.run_on_locked_transport_full("basic_publish", "Could not publish", request_id, move |channel, delivery_tag| {
          if channel.status.confirm() {
            if channel.acknowledgements.is_acked(delivery_tag) {
              Ok(Async::Ready(Some(delivery_tag)))
            } else if channel.acknowledgements.is_nacked(delivery_tag) {
              Ok(Async::Ready(None))
            } else {
              debug!("message with tag {} still in unacked", delivery_tag);
              task::current().notify();
              Ok(Async::NotReady)
            }
          } else {
            Ok(Async::Ready(None))
          }
        })
    }

    /// creates a consumer stream
    ///
    /// returns a future of a `Consumer` that resolves once the method succeeds
    ///
    /// `Consumer` implements `futures::Stream`, so it can be used with any of
    /// the usual combinators
    pub fn basic_consume(&self, queue: &Queue, consumer_tag: &str, options: BasicConsumeOptions, arguments: FieldTable) -> impl Future<Item = Consumer<T>, Error = Error> + Send + 'static {
        let consumer_tag = consumer_tag.to_string();
        let queue_name = queue.name();
        let mut consumer = Consumer::new(self.transport.clone(), self.id(), queue.name(), consumer_tag.to_owned());
        let subscriber = consumer.subscriber();
        let request_id = self.inner.basic_consume(&queue_name, &consumer_tag, options, arguments, Box::new(subscriber));
        let inner = self.inner.clone();

        self.run_on_locked_transport("basic_consume", "Could not start consumer", request_id).and_then(move |request_id| {
            future::poll_fn(move || {
              if let Some(consumer_tag) = inner.generated_names.get(request_id.expect("expected request_id")) {
                return Ok(Async::Ready(consumer_tag))
              } else {
                task::current().notify();
                return Ok(Async::NotReady)
              }
            })
          }).map(|consumer_tag| {
            trace!("basic_consume received response, returning consumer");
            consumer.update_consumer_tag(consumer_tag);
            consumer
        })
    }

    pub fn basic_cancel(&self, consumer_tag: &str, options: BasicCancelOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_cancel(consumer_tag, options);

        self.run_on_locked_transport("basic_cancel", "Could not cancel consumer", request_id).map(|_| ())
    }

    pub fn basic_recover(&self, options: BasicRecoverOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_recover(options);

        self.run_on_locked_transport("basic_recover", "Could not recover", request_id).map(|_| ())
    }

    pub fn basic_recover_async(&self, options: BasicRecoverAsyncOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_recover_async(options);

        self.run_on_locked_transport("basic_recover_async", "Could not recover", request_id).map(|_| ())
    }

    /// acks a message
    pub fn basic_ack(&self, delivery_tag: u64, multiple: bool) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_ack(delivery_tag, BasicAckOptions { multiple });

        self.run_on_locked_transport("basic_ack", "Could not ack message", request_id).map(|_| ())
    }

    /// nacks a message
    pub fn basic_nack(&self, delivery_tag: u64, multiple: bool, requeue: bool) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_nack(delivery_tag, BasicNackOptions { multiple, requeue });

        self.run_on_locked_transport("basic_nack", "Could not nack message", request_id).map(|_| ())
    }

    /// rejects a message
    pub fn basic_reject(&self, delivery_tag: u64, options: BasicRejectOptions) -> impl Future<Item = (), Error = Error> + Send + 'static {
        let request_id = self.inner.basic_reject(delivery_tag, options);

        self.run_on_locked_transport("basic_reject", "Could not reject message", request_id).map(|_| ())
    }
{'fraction_non_alphanumeric': 0.12910184609002281, 'fraction_numerical': 0.00074673304293715, 'mean_word_length': 2.961544782251438, 'pattern_counts': {'":': 0, '<': 74, '<?xml version=': 0, '>': 119, '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': '6681174', 'n_tokens_mistral': 6977, 'n_tokens_neox': 6741, 'n_words': 1811}