# coding=UTF-8
from __future__ import print_function, absolute_import, division

import logging
import time
import typing as tp
import warnings
from concurrent.futures import Future

import six

from coolamqp.attaches import Publisher, AttacheGroup, Consumer, Declarer
from coolamqp.attaches.utils import close_future
from coolamqp.clustering.events import ConnectionLost, MessageReceived, \
    NothingMuch, Event
from coolamqp.clustering.single import SingleNodeReconnector
from coolamqp.exceptions import ConnectionDead
from coolamqp.objects import Exchange, Message, Queue, QueueBind
from coolamqp.uplink import ListenerThread
from coolamqp.utils import monotonic

logger = logging.getLogger(__name__)

nothing_much = NothingMuch()


# If any spans are spawn here, it's Cluster's job to finish them, except for publish()
class Cluster(object):
    """
    Frontend for your AMQP needs.

    This has ListenerThread.

    Call .start() to connect to AMQP.

    It is not safe to fork() after .start() is called, but it's OK before.

    :param nodes: list of nodes, or a single node. For now, only one is supported.
    :param on_fail: callable/0 to call when connection fails in an
        unclean way. This is a one-shot
    :param extra_properties: refer to documentation in [/coolamqp/connection/connection.py]
        Connection.__init__
    :param log_frames: an object that supports logging each and every frame CoolAMQP sends and
        receives from the broker
    :type log_frames: tp.Optional[:class:`coolamqp.tracing.BaseFrameTracer`]
    :param name: name to appear in log items and prctl() for the listener thread
    :param on_blocked: callable to call when ConnectionBlocked/ConnectionUnblocked is received. It will be
        called with a value of True if connection becomes blocked, and False upon an unblock
    :param tracer: tracer, if opentracing is installed
    """

    # Events you can be informed about
    ST_LINK_LOST = 0  # Link has been lost
    ST_LINK_REGAINED = 1  # Link has been regained

    def __init__(self, nodes,  # type: tp.Union[NodeDefinition, tp.List[NodeDefinition]]
                 on_fail=None,  # type: tp.Optional[tp.Callable[[], None]]
                 extra_properties=None,
                 # type: tp.Optional[tp.List[tp.Tuple[bytes, tp.Tuple[bytes, str]]]]
                 log_frames=None,
                 name=None,  # type: tp.Optional[str]
                 on_blocked=None,  # type: tp.Callable[[bool], None],
                 tracer=None  # type: opentracing.Traccer
                 ):
        from coolamqp.objects import NodeDefinition
        if isinstance(nodes, NodeDefinition):
            nodes = [nodes]

        if len(nodes) > 1:
            raise NotImplementedError(u'Multiple nodes not supported yet')

        if tracer is not None:
            try:
                import opentracing
            except ImportError:
                raise RuntimeError('tracer given, but opentracing is not installed!')

        self.started = False            # type: bool
        self.tracer = tracer
        self.name = name or 'CoolAMQP'  # type: str
        self.node, = nodes              # type: NodeDefinition
        self.extra_properties = extra_properties
        self.log_frames = log_frames
        self.on_blocked = on_blocked    # type: tp.Optional[tp.Callable[[bool], None]]
        self.connected = False          # type: bool
        self.listener = None            # type: BaseListener
        self.attache_group = None       # type: AttacheGroup
        self.events = None              # type: six.moves.queue.Queue
        self.snr = None                 # type: SingleNodeReconnector
        self.pub_tr = None              # type: Publisher
        self.pub_na = None              # type: Publisher
        self.decl = None                # type: Declarer

        if on_fail is not None:
            def decorated():
                if not self.listener.terminating and self.connected:
                    on_fail()

            self.on_fail = decorated
        else:
            self.on_fail = None

    def bind(self, queue, exchange, routing_key, persistent=False, span=None,
             dont_trace=False):
        """
        Bind a queue to an exchange
        """
        if span is not None and not dont_trace:
            child_span = self._make_span('bind', span)
        else:
            child_span = None
        fut = self.decl.declare(QueueBind(queue, exchange, routing_key),
                                persistent=persistent,
                                span=child_span)
        return close_future(fut, child_span)

    def declare(self, obj,  # type: tp.Union[Queue, Exchange]
                persistent=False,  # type: bool
                span=None,  # type: tp.Optional[opentracing.Span]
                dont_trace=False    # type: bool
                ):  # type: (...) -> concurrent.futures.Future
        """
        Declare a Queue/Exchange

        :param obj: Queue/Exchange object
        :param persistent: should it be redefined upon reconnect?
        :param span: optional parent span, if opentracing is installed
        :param dont_trace: if True, a span won't be output
        :return: Future
        """
        if span is not None and not dont_trace:
            child_span = self._make_span('declare', span)
        else:
            child_span = None
        fut = self.decl.declare(obj, persistent=persistent, span=child_span)
        return close_future(fut, child_span)

    def drain(self, timeout, span=None, dont_trace=False):  # type: (float) -> Event
        """
        Return an Event.

        :param timeout: time to wait for an event. 0 means return immediately. None means block forever
        :param span: optional parent span, if opentracing is installed
        :param dont_trace: if True, this span won't be traced
        :return: an Event instance. NothingMuch is returned when there's nothing within a given timoeout
        """

        def fetch():
            try:
                if timeout == 0:
                    return self.events.get_nowait()
                else:
                    return self.events.get(True, timeout)
            except six.moves.queue.Empty:
                return nothing_much

        if span is not None and not dont_trace:
            from opentracing import tags
            parent_span = self.tracer.start_active_span('AMQP call',
                                                        child_of=span,
                                                        tags={
                                                            tags.SPAN_KIND: tags.SPAN_KIND_RPC_CLIENT,
                                                            tags.DATABASE_TYPE: 'amqp',
                                                            tags.DATABASE_STATEMENT: 'drain'
                                                        })

            with parent_span:
                return fetch()
        else:
            return fetch()

    def consume(self, queue, on_message=None, span=None,
                dont_trace=False,   # type: bool
                *args, **kwargs):
        # type: (Queue, tp.Callable[[MessageReceived], None]) -> tp.Tuple[Consumer, Future]
        """
        Start consuming from a queue.

        args and kwargs will be passed to Consumer constructor (coolamqp.attaches.consumer.Consumer).
        Don't use future_to_notify - it's done here!

        Take care not to lose the Consumer object - it's the only way to cancel a consumer!

        :param queue: Queue object, being consumed from right now.
            Note that name of anonymous queue might change at any time!
        :param on_message: callable that will process incoming messages
                           if you leave it at None, messages will be .put into self.events
        :param span: optional span, if opentracing is installed
        :param dont_trace: if True, this won't output a span
        :return: a tuple (Consumer instance, and a Future), that tells, when consumer is ready
        """
        if span is not None and not dont_trace:
            child_span = self._make_span('consume', span)
        else:
            child_span = None
        fut = Future()
        fut.set_running_or_notify_cancel()  # it's running right now
        on_message = on_message or (
            lambda msg: self.events.put_nowait(MessageReceived(msg)))
        con = Consumer(queue, on_message, future_to_notify=fut, span=span, *args,
                       **kwargs)
        self.attache_group.add(con)
        return con, close_future(fut, child_span)

    def delete_queue(self, queue):  # type: (coolamqp.objects.Queue) -> Future
        """
        Delete a queue.

        :param queue: Queue instance that represents what to delete
        :return: a Future (will succeed with None or fail with AMQPError)
        """
        return self.decl.delete_queue(queue)

    def _make_span(self, call, span):
        try:
            from opentracing import tags
        except ImportError:
            pass
        else:
            return self.tracer.start_span('AMQP call',
                                          child_of=span,
                                          tags={
                                              tags.SPAN_KIND: tags.SPAN_KIND_RPC_CLIENT,
                                              tags.DATABASE_TYPE: 'amqp',
                                              tags.DATABASE_STATEMENT: call
                                          })

    def publish(self, message,  # type: Message
                exchange=None,  # type: tp.Union[Exchange, str, bytes]
                routing_key=u'',  # type: tp.Union[str, bytes]
                tx=None,  # type: tp.Optional[bool]
                confirm=None,  # type: tp.Optional[bool]
                span=None,  # type: tp.Optional[opentracing.Span]
                dont_trace=False    # type: bool
                ):  # type: (...) -> tp.Optional[Future]
        """
        Publish a message.

        :param message: Message to publish
        :param exchange: exchange to use. Default is the "direct" empty-name exchange.
        :param routing_key: routing key to use
        :param confirm: Whether to publish it using confirms/transactions.
                        If you choose so, you will receive a Future that can be used
                        to check it broker took responsibility for this message.
                        Note that if tx if False, and message cannot be delivered to broker at once,
                        it will be discarded
        :param tx: deprecated, alias for confirm
        :param span: optionally, current span, if opentracing is installed
        :param dont_trace: if set to True, a span won't be generated
        :return: Future to be finished on completion or None, is confirm/tx was not chosen
        """
        if self.tracer is not None and not dont_trace:
            span = self._make_span('publish', span)

        if isinstance(exchange, Exchange):
            exchange = exchange.name.encode('utf8')
        elif exchange is None:
            exchange = b''
        elif isinstance(exchange, six.text_type):
            exchange = exchange.encode('utf8')

        if isinstance(routing_key, six.text_type):
            routing_key = routing_key.encode('utf8')

        if tx is not None:  # confirm is a drop-in replacement. tx is unfortunately named
            warnings.warn(u'Use confirm kwarg instead', DeprecationWarning)

            if confirm is not None:
                raise RuntimeError(
                    u'Using both tx= and confirm= at once does not make sense')
        elif confirm is not None:
            tx = confirm
        else:
            tx = False

        try:
            if tx:
                clb = self.pub_tr
            else:
                clb = self.pub_na
            return clb.publish(message, exchange, routing_key, span)
        except Publisher.UnusablePublisher:
            raise NotImplementedError(
                u'Sorry, this functionality is not yet implemented!')

    def start(self, wait=True, timeout=10.0):  # type: (bool, float, bool) -> None
        """
        Connect to broker. Initialize Cluster.

        Only after this call is Cluster usable.
        It is not safe to fork after this.

        :param wait: block until connection is ready
        :param timeout: timeout to wait until the connection is ready. If it is not, a
                        ConnectionDead error will be raised
        :raise RuntimeError: called more than once
        :raise ConnectionDead: failed to connect within timeout
        """
        if self.started:
            raise RuntimeError(u'[%s] This was already called!' % (self.name,))
        self.started = True

        self.listener = ListenerThread(name=self.name)

        self.attache_group = AttacheGroup()

        self.events = six.moves.queue.Queue()  # for coolamqp.clustering.events.*

        self.snr = SingleNodeReconnector(self.node, self.attache_group,
                                         self.listener, self.extra_properties,
                                         self.log_frames, self.name)
        self.snr.on_fail.add(lambda: self.events.put_nowait(ConnectionLost()))
        if self.on_fail is not None:
            self.snr.on_fail.add(self.on_fail)

        if self.on_blocked is not None:
            self.snr.on_blocked.add(self.on_blocked)

        # Spawn a transactional publisher and a noack publisher
        self.pub_tr = Publisher(Publisher.MODE_CNPUB, self)
        self.pub_na = Publisher(Publisher.MODE_NOACK, self)
        self.decl = Declarer(self)

        self.attache_group.add(self.pub_tr)
        self.attache_group.add(self.pub_na)
        self.attache_group.add(self.decl)

        self.listener.init()
        self.listener.start()
        self.snr.connect(timeout=timeout)

        if wait:
            # this is only going to take a short amount of time, so we're fine with polling
            start_at = monotonic()
            while not self.connected and monotonic() - start_at < timeout:
                time.sleep(0.1)
            if not self.connected:
                raise ConnectionDead(
                    '[%s] Could not connect within %s seconds' % (self.name, timeout,))

    def shutdown(self, wait=True):  # type: (bool) -> None
        """
        Terminate all connections, release resources - finish the job.

        :param wait: block until this is done
        :raise RuntimeError: if called without start() being called first
        """
        self.connected = False
        if not self.started:
            raise RuntimeError(u'shutdown without start')

        logger.info('[%s] Commencing shutdown', self.name)

        self.listener.terminate()
        if wait:
            self.listener.join()
