from amqplib import client_0_8 as amqp

try:
    # cjson is the fastest
    import cjson
    serialize = cjson.encode
    deserialize = cjson.decode
except ImportError:
    try:
        # Then try to find the latest version of simplejson.
        # Later versions has C speedups which makes it pretty fast.
        import simplejson
        serialize = simplejson.dumps
        deserialize = simplejson.loads
    except ImportError:
        # If all of the above fails, fallback to the simplejson
        # embedded in Django.
        from django.utils import simplejson
        serialize = simplejson.dumps
        deserialize = simplejson.loads


class Message(object):
    """Wrapper around amqplib.client_0_8.Message."""
    def __init__(self, amqp_message, channel):
        self.amqp_message = amqp_message
        self.channel = channel

    def ack(self):
        """Acknowledge this message as being processed.,
        
        This will remove the message from the queue."""
        return self.channel.basic_ack(self.delivery_tag)

    def reject(self):
        """Reject this message.
        The message will then be discarded by the server.
        """
        return self.channel.basic_reject(self.delivery_tag, requeue=False)

    def requeue(self):
        """Reject this message and put it back on the queue.

        You must not use this method as a means of selecting messages
        to process."""
        return self.channel.basic_reject(self.delivery_tag, requeue=True)

    @property
    def body(self):
        return self.amqp_message.body

    @property
    def delivery_tag(self):
        return self.amqp_message.delivery_tag


class Consumer(object):
    queue = ""
    exchange = ""
    routing_key = ""
    durable = True
    exclusive = False
    auto_delete = False
    exchange_type = "direct"
    channel_open = False

    def __init__(self, connection, queue=None, exchange=None, routing_key=None,
            **kwargs):
        self.connection = connection()
        self.queue = queue or self.queue

        self.exchange = exchange or self.exchange
        self.routing_key = routing_key or self.routing_key
        self.decoder = kwargs.get("decoder", deserialize)
        self.durable = kwargs.get("durable", self.durable)
        self.exclusive = kwargs.get("exclusive", self.exclusive)
        self.auto_delete = kwargs.get("auto_delete", self.auto_delete)
        self.exchange_type = kwargs.get("exchange_type", self.exchange_type)
        self.channel = self.build_channel()

    def build_channel(self):
        channel = self.connection.connection.channel()
        if self.queue:
            channel.queue_declare(queue=self.queue, durable=self.durable,
                                  exclusive=self.exclusive,
                                  auto_delete=self.auto_delete)
        if self.exchange:
            channel.exchange_declare(exchange=self.exchange,
                                     type=self.exchange_type,
                                     durable=self.durable,
                                     auto_delete=self.auto_delete)
        if self.queue:
            channel.queue_bind(queue=self.queue, exchange=self.exchange,
                               routing_key=self.routing_key)
        return channel

    def receive_callback(self, message):
        message_data = self.decoder(message.body)
        self.receive(message_data, message)

    def receive(self, message_data, message):
        raise NotImplementedError(
                "Consumers must implement the receive method")

    def fetch(self):
        if not self.channel.connection:
            self.channel = self.build_channel()
        raw_message = self.channel.basic_get(self.queue)
        if not raw_message:
            return None
        return Message(raw_message, channel=self.channel)

    def process_next(self):
        message = self.fetch()
        if message:
            self.receive_callback(message)
            message.ack()
        return message

    def discard_all(self):
        """Discard all waiting messages.
       
        Returns the number of messages discarded.
        *WARNING*: All incoming messages will be ignored and not processed.
        """
        discarded_count = 0
        while True:
            message = self.fetch()
            if message is None:
                return discarded_count
            message.ack()
            discarded_count = discarded_count + 1

    def next(self):
        raise DeprecationWarning(
                "next() is deprecated, use process_next() instead.")
        return self.process_next()

    def wait(self):
        if not self.channel.connection:
            self.channel = self.build_channel()
        self.channel_open = True
        self.channel.basic_consume(queue=self.queue, no_ack=True,
                callback=self.receive_callback,
                consumer_tag=self.__class__.__name__)
        yield self.channel.wait()

    def iterqueue(self, limit=None):
        for items_since_start in itertools.count():
            item = self.next()
            if item is None or (limit and items_since_start > limit):
                raise StopIteration
            yield item

    def close(self):
        if self.channel_open:
            self.channel.basic_cancel(self.__class__.__name__)
        if getattr(self, "channel") and self.channel.is_open:
            self.channel.close()


class Publisher(object):
    exchange = ""
    routing_key = ""
    delivery_mode = 2 # Persistent

    def __init__(self, connection, exchange=None, routing_key=None, **kwargs):
        self.connection = connection()
        self.exchange = exchange or self.exchange
        self.routing_key = routing_key or self.routing_key
        self.encoder = kwargs.get("encoder", serialize)
        self.delivery_mode = kwargs.get("delivery_mode", self.delivery_mode)
        self.channel = self.build_channel()

    def build_channel(self):
        return self.connection.connection.channel()

    def create_message(self, message_data):
        # Recreate channel if connection lost.
        if not self.channel.connection:
            self.channel = self.build_channel()

        message_data = self.encoder(message_data)
        message = amqp.Message(message_data)
        message.properties["delivery_mode"] = self.delivery_mode
        return message

    def send(self, message_data, delivery_mode=None):
        message = self.create_message(message_data)
        self.channel.basic_publish(message, exchange=self.exchange,
                                              routing_key=self.routing_key)

    def close(self):
        if getattr(self, "channel") and self.channel.is_open:
            self.channel.close()


class Messaging(object):
    queue = ""
    exchange = ""
    routing_key = ""
    publisher_cls = Publisher
    consumer_cls = Consumer

    def __init__(self, connection_cls, **kwargs):
        self.connection_cls = connection_cls
        self.exchange = kwargs.get("exchange", self.exchange)
        self.queue = kwargs.get("queue", self.queue)
        self.routing_key = kwargs.get("routing_key", self.routing_key)
        self.publisher = self.publisher_cls(connection_cls,
                exchange=self.exchange, routing_key=self.routing_key)
        self.consumer = self.consumer_cls(connection_cls, queue=self.queue,
                exchange=self.exchange, routing_key=self.routing_key)
        self.consumer.receive = self.receive_callback

    def send(self, message_data, delivery_mode=None):
        self.publisher.send(message_data, delivery_mode=delivery_mode)

    def receive_callback(self, message_data, message):
        self.receive(message_data, message)

    def receive(self, message_data, message):
        raise NotImplementedError(
                "Messaging classes must implement the receive method")

    def next(self):
        return self.consumer.next()

    def close(self):
        self.consumer.close()
        self.publisher.close()

@@ -31,16 +31,32 @@
  as amqp
+%0Aimport warnings
 %0A%0Atry:%0A 
@@ -4503,50 +4503,22 @@
     
-raise DeprecationWarning(%0A                
+warnings.warn(
 %22nex
@@ -4564,16 +4564,52 @@
 nstead.%22
+,%0A                DeprecationWarning
 )%0A      

Use warnings.warn with DeprecationWarning instead of raise