private static void rate_limit_notify(IBasicConsumer consumer,
                                       BasicDeliverEventArgs eargs) {
     
     string[] EMAIL_RECIPS = new string[] {"[email protected]"};
     
     IBasicProperties msg_props = eargs.BasicProperties;
     String msg_body = Encoding.ASCII.GetString(eargs.Body);
     
     //#/(ascdn.4) Decode our message from JSON
     msg_body = JsonConvert.DeserializeObject
                            <string>(msg_body);
     
     //#/(ascdn.5) Transmit e-mail to SMTP server
     send_mail(EMAIL_RECIPS,
               "RATE LIMIT ALERT!",
               msg_body);
     
     Console.WriteLine("Sent alert via e-mail! Alert Text: " +
                       msg_body + " Recipients: " +
                       string.Join(",", EMAIL_RECIPS));
     
     //#/(ascdn.6) Acknowledge the message
     consumer.Model.BasicAck(eargs.DeliveryTag,
                             false);
 }

 /// <summary>Start a Basic content-class consumer.</summary>
 public static string BasicConsume(this IModel model, string queue,
     bool autoAck,
     string consumerTag,
     IBasicConsumer consumer)
 {
     return model.BasicConsume(queue, autoAck, consumerTag, false, false, null, consumer);
 }

        public virtual void HandleMessage(IBasicConsumer consumer, BasicDeliverEventArgs eventArg)
        {
            var completionTask = _jobFactory(eventArg);
            completionTask.ContinueWith(task =>
            {
                try
                {
                    if (task.IsFaulted)
                    {
                        HandleError(consumer, eventArg, task.Exception);
                    }
                }
                catch (Exception ex)
                {
                    _watcher.Error(ex);
                }
                finally
                {
                    // Broadcast msgs
                    AfterHandlingMessage(consumer, eventArg);

                    //NOTE: Only this way, the new event to interupt other consumers will override previous resume event
                    if (HandlingComplete != null)
                    {
                        // Release pool + DoAck
                        HandlingComplete(eventArg);
                    }
                }
            });
        }

 public void HandleBasicDeliver(IBasicConsumer consumer,
                                string consumerTag,
                                ulong deliveryTag,
                                bool redelivered,
                                string exchange,
                                string routingKey,
                                IBasicProperties basicProperties,
                                byte[] body)
 {
     UnlessShuttingDown(() =>
     {
         try
         {
             consumer.HandleBasicDeliver(consumerTag,
                                         deliveryTag,
                                         redelivered,
                                         exchange,
                                         routingKey,
                                         basicProperties,
                                         body);
         }
         catch (Exception e)
         {
             var details = new Dictionary<string, object>()
             {
                 {"consumer", consumer},
                 {"context",  "HandleBasicDeliver"}
             };
             model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
         }
     });
 }

        public SubscriberBase(IModel model, string queue, IBasicConsumer consumer, SubscriberConfigurator configurator)
        {
            _model = model;

            _queue = queue;

            _consumer = consumer;
            _configurator = configurator;
        }

        private void ConsumerOnReceived(IBasicConsumer sender, BasicDeliverEventArgs args)
        {
            var millisecondsTimeout = m_rnd.Next(1, 10) * 1000;
            Console.WriteLine("Message received: {0}, sleeping for {1} seconds", Encoding.UTF8.GetString(args.Body), millisecondsTimeout/1000);

            Thread.Sleep(millisecondsTimeout);

            sender.Model.BasicAck(args.DeliveryTag, false);
        }

        private static void ConsumerOnReceived(IBasicConsumer sender, BasicDeliverEventArgs args)
        {
            var n = int.Parse(Encoding.UTF8.GetString(args.Body));

            Console.WriteLine("Received computation request for n = {0}, correlation id {1}", n, args.BasicProperties.CorrelationId);

            var fib = Fibonacci(n);

            var properties = sender.Model.CreateBasicProperties();
            properties.CorrelationId = args.BasicProperties.CorrelationId;

            sender.Model.BasicPublish("", args.BasicProperties.ReplyTo, properties, Encoding.UTF8.GetBytes(fib.ToString()));

            sender.Model.BasicAck(args.DeliveryTag, false);
        }

        public ConsumerExecutionContext(
            Func<byte[], MessageProperties, MessageReceivedInfo, Task> userHandler, 
            MessageReceivedInfo info, 
            MessageProperties properties, 
            byte[] body, 
            IBasicConsumer consumer)
        {
            Preconditions.CheckNotNull(userHandler, "userHandler");
            Preconditions.CheckNotNull(info, "info");
            Preconditions.CheckNotNull(properties, "properties");
            Preconditions.CheckNotNull(body, "body");
            Preconditions.CheckNotNull(consumer, "consumer");

            UserHandler = userHandler;
            Info = info;
            Properties = properties;
            Body = body;
            Consumer = consumer;
        }

 public void HandleBasicCancelOk(IBasicConsumer consumer, string consumerTag)
 {
     UnlessShuttingDown(() =>
     {
         try
         {
             consumer.HandleBasicCancelOk(consumerTag);
         }
         catch (Exception e)
         {
             var details = new Dictionary<string, object>()
             {
                 {"consumer", consumer},
                 {"context",  "HandleBasicCancelOk"}
             };
             model.OnCallbackException(CallbackExceptionEventArgs.Build(e, details));
         }
     });
 }

 public void DoAckForTest(BasicDeliverEventArgs basicDeliverEventArgs, IBasicConsumer subscriptionInfo)
 {
     DoAck(basicDeliverEventArgs, subscriptionInfo);
 }

 internal protected override void DoAck(BasicDeliverEventArgs basicDeliverEventArgs, IBasicConsumer subscriptionInfo)
 {
     base.DoAck(basicDeliverEventArgs, subscriptionInfo);
     WaitHandler.Set();
 }

 private void HandleEvent(IBasicConsumer sender, BasicDeliverEventArgs args)
 {
     var elementBody = Encoding.Unicode.GetString(args.Body);
     var element = XElement.Parse(elementBody);
     DecodeAndRouteEvent(element);
 }

        protected void TestExceptionHandlingWith(IBasicConsumer consumer, FailingOp fn)
        {
            Object o = new Object();
            bool notified = false;
            string q = Model.QueueDeclare();


            Model.CallbackException += (m, evt) => {
                notified = true;
                Monitor.PulseAll(o);
            };

            string tag = Model.BasicConsume(q, true, consumer);
            fn(Model, q, consumer, tag);
            WaitOn(o);

            Assert.IsTrue(notified);
        }

        public string BasicConsume(string queue,
                                   bool noAck,
                                   string consumerTag,
                                   bool noLocal,
                                   bool exclusive,
                                   IDictionary<string, object> arguments,
                                   IBasicConsumer consumer)
        {
            ModelShutdown += new ModelShutdownEventHandler(consumer.HandleModelShutdown);

            BasicConsumerRpcContinuation k = new BasicConsumerRpcContinuation();
            k.m_consumer = consumer;

            Enqueue(k);
            // Non-nowait. We have an unconventional means of getting
            // the RPC response, but a response is still expected.
            _Private_BasicConsume(queue, consumerTag, noLocal, noAck, exclusive,
                                  /*nowait:*/ false, arguments);
            k.GetReply();
            string actualConsumerTag = k.m_consumerTag;

            return actualConsumerTag;
        }

 public string BasicConsume(string queue, bool noAck, IDictionary filter, IBasicConsumer consumer)
 {
     throw new NotImplementedException();
 }

 public string BasicConsume(string queue, bool noAck, string consumerTag, bool noLocal, bool exclusive, IDictionary<string, object> arguments,
     IBasicConsumer consumer)
 {
     throw new NotImplementedException();
 }

 Task<string> ModelContext.BasicConsume(string queue, bool noAck, IBasicConsumer consumer)
 {
     return Task.Factory.StartNew(() => _model.BasicConsume(queue, noAck, consumer),
         _participant.StoppedToken, TaskCreationOptions.None, _taskScheduler);
 }

        internal protected override void DoAck(BasicDeliverEventArgs basicDeliverEventArgs, IBasicConsumer subscriptionInfo)
        {
            if (IsDisposed)
            {
                return;
            }

            _subscription.Ack(basicDeliverEventArgs.ConsumerTag, basicDeliverEventArgs.DeliveryTag);
        }

 public SubscriptionInfo(IBasicConsumer consumer, MessageCallback callback)
 {
     Consumer = consumer;
     Callback = callback;
 }

 public string BasicConsume(string queue,
     bool noAck,
     string consumerTag,
     bool noLocal,
     bool exclusive,
     IDictionary<string, object> arguments,
     IBasicConsumer consumer)
 {
     var result = m_delegate.BasicConsume(queue, noAck, consumerTag, noLocal,
         exclusive, arguments, consumer);
     RecordedConsumer rc = new RecordedConsumer(this, queue).
         WithConsumerTag(result).
         WithConsumer(consumer).
         WithExclusive(exclusive).
         WithAutoAck(noAck).
         WithArguments(arguments);
     m_connection.RecordConsumer(result, rc);
     return result;
 }