"""
Created on Fri Aug 3 17:00:44 2018

@author: Bge
"""
import pika,json,logging
class RabbitMQClient:
  def __init__(self, conn_str='amqp://user:pwd@host:port/%2F'):
    self.exchange_type = "direct"
    self.connection_string = conn_str
    self.connection = pika.BlockingConnection(pika.URLParameters(self.connection_string))
    self.channel = self.connection.channel()
    self._declare_retry_queue() #RetryQueue and RetryExchange
    logging.debug("connection established")
  def close_connection(self):
    self.connection.close()
    logging.debug("connection closed")
  def declare_exchange(self, exchange):
    self.channel.exchange_declare(exchange=exchange,
                   exchange_type=self.exchange_type,
                   durable=True)
  def declare_queue(self, queue):
    self.channel.queue_declare(queue=queue,
                  durable=True,)
  def declare_delay_queue(self, queue,DLX='RetryExchange',TTL=60000):
    """
    創(chuàng)建延遲隊(duì)列
    :param TTL: ttl的單位是us，ttl=60000 表示 60s
    :param queue:
    :param DLX:死信轉(zhuǎn)發(fā)的exchange
    :return:
    """
    arguments={}
    if DLX:
      #設(shè)置死信轉(zhuǎn)發(fā)的exchange
      arguments[ 'x-dead-letter-exchange']=DLX
    if TTL:
      arguments['x-message-ttl']=TTL
    print(arguments)
    self.channel.queue_declare(queue=queue,
                  durable=True,
                  arguments=arguments)
  def _declare_retry_queue(self):
    """
    創(chuàng)建異常交換器和隊(duì)列，用于存放沒(méi)有正常處理的消息。
    :return:
    """
    self.channel.exchange_declare(exchange='RetryExchange',
                   exchange_type='fanout',
                   durable=True)
    self.channel.queue_declare(queue='RetryQueue',
                  durable=True)
    self.channel.queue_bind('RetryQueue', 'RetryExchange','RetryQueue')
  def publish_message(self,routing_key, msg,exchange='',delay=0,TTL=None):
    """
    發(fā)送消息到指定的交換器
    :param exchange: RabbitMQ交換器
    :param msg: 消息實(shí)體，是一個(gè)序列化的JSON字符串
    :return:
    """
    if delay==0:
      self.declare_queue(routing_key)
    else:
      self.declare_delay_queue(routing_key,TTL=TTL)
    if exchange!='':
      self.declare_exchange(exchange)
    self.channel.basic_publish(exchange=exchange,
                  routing_key=routing_key,
                  body=msg,
                  properties=pika.BasicProperties(
                    delivery_mode=2,
                    type=exchange
                  ))
    self.close_connection()
    print("message send out to %s" % exchange)
    logging.debug("message send out to %s" % exchange)
  def start_consume(self,callback,queue='#',delay=1):
    """
    啟動(dòng)消費(fèi)者，開(kāi)始消費(fèi)RabbitMQ中的消息
    :return:
    """
    if delay==1:
      queue='RetryQueue'
    else:
      self.declare_queue(queue)
    self.channel.basic_qos(prefetch_count=1)
    try:
      self.channel.basic_consume( # 消費(fèi)消息
        callback, # 如果收到消息，就調(diào)用callback函數(shù)來(lái)處理消息
        queue=queue, # 你要從那個(gè)隊(duì)列里收消息
      )
      self.channel.start_consuming()
    except KeyboardInterrupt:
      self.stop_consuming()
  def stop_consuming(self):
    self.channel.stop_consuming()
    self.close_connection()
  def message_handle_successfully(channel, method):
    """
    如果消息處理正常完成，必須調(diào)用此方法，
    否則RabbitMQ會(huì)認(rèn)為消息處理不成功，重新將消息放回待執(zhí)行隊(duì)列中
    :param channel: 回調(diào)函數(shù)的channel參數(shù)
    :param method: 回調(diào)函數(shù)的method參數(shù)
    :return:
    """
    channel.basic_ack(delivery_tag=method.delivery_tag)
  def message_handle_failed(channel, method):
    """
    如果消息處理失敗，應(yīng)該調(diào)用此方法，會(huì)自動(dòng)將消息放入異常隊(duì)列
    :param channel: 回調(diào)函數(shù)的channel參數(shù)
    :param method: 回調(diào)函數(shù)的method參數(shù)
    :return:
    """
    channel.basic_reject(delivery_tag=method.delivery_tag, requeue=False)

from MQ.RabbitMQ import RabbitMQClient
print("start program")
client = RabbitMQClient()
msg1 = '{"key":"value"}'
client.publish_message('test-delay',msg1,delay=1,TTL=10000)
print("message send out")

from MQ.RabbitMQ import RabbitMQClient
import json
print("start program")
client = RabbitMQClient()
def callback(ch, method, properties, body):
    msg = body.decode()
    print(msg)
    # 如果處理成功，則調(diào)用此消息回復(fù)ack，表示消息成功處理完成。
    RabbitMQClient.message_handle_successfully(ch, method)
queue_name = "RetryQueue"
client.start_consume(callback,queue_name,delay=0)