#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#

from __future__ import absolute_import

import collections
import time
import threading

from cproton import pn_reactor_collector, pn_collector_release

from ._exceptions import ProtonException, ConnectionException, LinkException, Timeout
from ._delivery import Delivery
from ._endpoints import Endpoint, Link
from ._events import Handler
from ._url import Url

from ._reactor import Container
from ._handlers import MessagingHandler, IncomingMessageHandler


class BlockingLink(object):
    def __init__(self, connection, link):
        self.connection = connection
        self.link = link
        self.connection.wait(lambda: not (self.link.state & Endpoint.REMOTE_UNINIT),
                             msg="Opening link %s" % link.name)
        self._checkClosed()

    def _waitForClose(self, timeout=1):
        try:
            self.connection.wait(lambda: self.link.state & Endpoint.REMOTE_CLOSED,
                                 timeout=timeout,
                                 msg="Opening link %s" % self.link.name)
        except Timeout as e:
            pass
        self._checkClosed()

    def _checkClosed(self):
        if self.link.state & Endpoint.REMOTE_CLOSED:
            self.link.close()
            if not self.connection.closing:
                raise LinkDetached(self.link)

    def close(self):
        self.link.close()
        self.connection.wait(lambda: not (self.link.state & Endpoint.REMOTE_ACTIVE),
                             msg="Closing link %s" % self.link.name)

    # Access to other link attributes.
    def __getattr__(self, name):
        return getattr(self.link, name)


class SendException(ProtonException):
    """
    Exception used to indicate an exceptional state/condition on a send request
    """

    def __init__(self, state):
        self.state = state


def _is_settled(delivery):
    return delivery.settled or delivery.link.snd_settle_mode == Link.SND_SETTLED


class BlockingSender(BlockingLink):
    def __init__(self, connection, sender):
        super(BlockingSender, self).__init__(connection, sender)
        if self.link.target and self.link.target.address and self.link.target.address != self.link.remote_target.address:
            # this may be followed by a detach, which may contain an error condition, so wait a little...
            self._waitForClose()
            # ...but close ourselves if peer does not
            self.link.close()
            raise LinkException("Failed to open sender %s, target does not match" % self.link.name)

    def send(self, msg, timeout=False, error_states=None):
        delivery = self.link.send(msg)
        self.connection.wait(lambda: _is_settled(delivery), msg="Sending on sender %s" % self.link.name,
                             timeout=timeout)
        if delivery.link.snd_settle_mode != Link.SND_SETTLED:
            delivery.settle()
        bad = error_states
        if bad is None:
            bad = [Delivery.REJECTED, Delivery.RELEASED]
        if delivery.remote_state in bad:
            raise SendException(delivery.remote_state)
        return delivery


class Fetcher(MessagingHandler):
    def __init__(self, connection, prefetch):
        super(Fetcher, self).__init__(prefetch=prefetch, auto_accept=False)
        self.connection = connection
        self.incoming = collections.deque([])
        self.unsettled = collections.deque([])

    def on_message(self, event):
        self.incoming.append((event.message, event.delivery))
        self.connection.container.yield_()  # Wake up the wait() loop to handle the message.

    def on_link_error(self, event):
        if event.link.state & Endpoint.LOCAL_ACTIVE:
            event.link.close()
            if not self.connection.closing:
                raise LinkDetached(event.link)

    def on_connection_error(self, event):
        if not self.connection.closing:
            raise ConnectionClosed(event.connection)

    @property
    def has_message(self):
        return len(self.incoming)

    def pop(self):
        message, delivery = self.incoming.popleft()
        if not delivery.settled:
            self.unsettled.append(delivery)
        return message

    def settle(self, state=None):
        delivery = self.unsettled.popleft()
        if state:
            delivery.update(state)
        delivery.settle()


class BlockingReceiver(BlockingLink):
    def __init__(self, connection, receiver, fetcher, credit=1):
        super(BlockingReceiver, self).__init__(connection, receiver)
        if self.link.source and self.link.source.address and self.link.source.address != self.link.remote_source.address:
            # this may be followed by a detach, which may contain an error condition, so wait a little...
            self._waitForClose()
            # ...but close ourselves if peer does not
            self.link.close()
            raise LinkException("Failed to open receiver %s, source does not match" % self.link.name)
        if credit: receiver.flow(credit)
        self.fetcher = fetcher
        self.container = connection.container

    def __del__(self):
        self.fetcher = None
        # The next line causes a core dump if the Proton-C reactor finalizes
        # first.  The self.container reference prevents out of order reactor
        # finalization. It may not be set if exception in BlockingLink.__init__
        if hasattr(self, "container"):
            self.link.handler = None  # implicit call to reactor

    def receive(self, timeout=False):
        if not self.fetcher:
            raise Exception("Can't call receive on this receiver as a handler was provided")
        if not self.link.credit:
            self.link.flow(1)
        self.connection.wait(lambda: self.fetcher.has_message, msg="Receiving on receiver %s" % self.link.name,
                             timeout=timeout)
        return self.fetcher.pop()

    def accept(self):
        self.settle(Delivery.ACCEPTED)

    def reject(self):
        self.settle(Delivery.REJECTED)

    def release(self, delivered=True):
        if delivered:
            self.settle(Delivery.MODIFIED)
        else:
            self.settle(Delivery.RELEASED)

    def settle(self, state=None):
        if not self.fetcher:
            raise Exception("Can't call accept/reject etc on this receiver as a handler was provided")
        self.fetcher.settle(state)


class LinkDetached(LinkException):
    def __init__(self, link):
        self.link = link
        if link.is_sender:
            txt = "sender %s to %s closed" % (link.name, link.target.address)
        else:
            txt = "receiver %s from %s closed" % (link.name, link.source.address)
        if link.remote_condition:
            txt += " due to: %s" % link.remote_condition
            self.condition = link.remote_condition.name
        else:
            txt += " by peer"
            self.condition = None
        super(LinkDetached, self).__init__(txt)


class ConnectionClosed(ConnectionException):
    def __init__(self, connection):
        self.connection = connection
        txt = "Connection %s closed" % connection.hostname
        if connection.remote_condition:
            txt += " due to: %s" % connection.remote_condition
            self.condition = connection.remote_condition.name
        else:
            txt += " by peer"
            self.condition = None
        super(ConnectionClosed, self).__init__(txt)


class BlockingConnection(Handler):
    """
    A synchronous style connection wrapper.

    This object's implementation uses OS resources.  To ensure they
    are released when the object is no longer in use, make sure that
    object operations are enclosed in a try block and that close() is
    always executed on exit.
    """

    def __init__(self, url, timeout=None, container=None, ssl_domain=None, heartbeat=None, **kwargs):
        self.disconnected = False
        self.timeout = timeout or 60
        self.container = container or Container()
        self.container.timeout = self.timeout
        self.container.start()
        self.url = Url(url).defaults()
        self.conn = None
        self.closing = False
        failed = True
        try:
            self.conn = self.container.connect(url=self.url, handler=self, ssl_domain=ssl_domain, reconnect=False,
                                               heartbeat=heartbeat, **kwargs)
            self.wait(lambda: not (self.conn.state & Endpoint.REMOTE_UNINIT),
                      msg="Opening connection")
            failed = False
        finally:
            if failed and self.conn:
                self.close()

    def create_sender(self, address, handler=None, name=None, options=None):
        return BlockingSender(self, self.container.create_sender(self.conn, address, name=name, handler=handler,
                                                                 options=options))

    def create_receiver(self, address, credit=None, dynamic=False, handler=None, name=None, options=None):
        prefetch = credit
        if handler:
            fetcher = None
            if prefetch is None:
                prefetch = 1
        else:
            fetcher = Fetcher(self, credit)
        return BlockingReceiver(
            self,
            self.container.create_receiver(self.conn, address, name=name, dynamic=dynamic, handler=handler or fetcher,
                                           options=options), fetcher, credit=prefetch)

    def close(self):
        # TODO: provide stronger interrupt protection on cleanup.  See PEP 419
        if self.closing:
            return
        self.closing = True
        self.container.errors = []
        try:
            if self.conn:
                self.conn.close()
                self.wait(lambda: not (self.conn.state & Endpoint.REMOTE_ACTIVE),
                          msg="Closing connection")
        finally:
            self.conn.free()
            # Nothing left to block on.  Allow reactor to clean up.
            self.run()
            self.conn = None
            self.container.global_handler = None  # break circular ref: container to cadapter.on_error
            pn_collector_release(pn_reactor_collector(self.container._impl))  # straggling event may keep reactor alive
            self.container = None

    def _is_closed(self):
        return self.conn.state & (Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED)

    def run(self):
        """ Hand control over to the event loop (e.g. if waiting indefinitely for incoming messages) """
        while self.container.process(): pass
        self.container.stop()
        self.container.process()

    def wait(self, condition, timeout=False, msg=None):
        """Call process until condition() is true"""
        if timeout is False:
            timeout = self.timeout
        if timeout is None:
            while not condition() and not self.disconnected:
                self.container.process()
        else:
            container_timeout = self.container.timeout
            self.container.timeout = timeout
            try:
                deadline = time.time() + timeout
                while not condition() and not self.disconnected:
                    self.container.process()
                    if deadline < time.time():
                        txt = "Connection %s timed out" % self.url
                        if msg: txt += ": " + msg
                        raise Timeout(txt)
            finally:
                self.container.timeout = container_timeout
        if self.disconnected or self._is_closed():
            self.container.stop()
            self.conn.handler = None  # break cyclical reference
        if self.disconnected and not self._is_closed():
            raise ConnectionException(
                "Connection %s disconnected: %s" % (self.url, self.disconnected))

    def on_link_remote_close(self, event):
        if event.link.state & Endpoint.LOCAL_ACTIVE:
            event.link.close()
            if not self.closing:
                raise LinkDetached(event.link)

    def on_connection_remote_close(self, event):
        if event.connection.state & Endpoint.LOCAL_ACTIVE:
            event.connection.close()
            if not self.closing:
                raise ConnectionClosed(event.connection)

    def on_transport_tail_closed(self, event):
        self.on_transport_closed(event)

    def on_transport_head_closed(self, event):
        self.on_transport_closed(event)

    def on_transport_closed(self, event):
        self.disconnected = event.transport.condition or "unknown"


class AtomicCount(object):
    def __init__(self, start=0, step=1):
        """Thread-safe atomic counter. Start at start, increment by step."""
        self.count, self.step = start, step
        self.lock = threading.Lock()

    def next(self):
        """Get the next value"""
        self.lock.acquire()
        self.count += self.step;
        result = self.count
        self.lock.release()
        return result


class SyncRequestResponse(IncomingMessageHandler):
    """
    Implementation of the synchronous request-response (aka RPC) pattern.
    @ivar address: Address for all requests, may be None.
    @ivar connection: Connection for requests and responses.
    """

    correlation_id = AtomicCount()

    def __init__(self, connection, address=None):
        """
        Send requests and receive responses. A single instance can send many requests
        to the same or different addresses.

        @param connection: A L{BlockingConnection}
        @param address: Address for all requests.
            If not specified, each request must have the address property set.
            Successive messages may have different addresses.
        """
        super(SyncRequestResponse, self).__init__()
        self.connection = connection
        self.address = address
        self.sender = self.connection.create_sender(self.address)
        # dynamic=true generates a unique address dynamically for this receiver.
        # credit=1 because we want to receive 1 response message initially.
        self.receiver = self.connection.create_receiver(None, dynamic=True, credit=1, handler=self)
        self.response = None

    def call(self, request):
        """
        Send a request message, wait for and return the response message.

        @param request: A L{proton.Message}. If L{self.address} is not set the 
            L{self.address} must be set and will be used.
        """
        if not self.address and not request.address:
            raise ValueError("Request message has no address: %s" % request)
        request.reply_to = self.reply_to
        request.correlation_id = correlation_id = str(self.correlation_id.next())
        self.sender.send(request)

        def wakeup():
            return self.response and (self.response.correlation_id == correlation_id)

        self.connection.wait(wakeup, msg="Waiting for response")
        response = self.response
        self.response = None  # Ready for next response.
        self.receiver.flow(1)  # Set up credit for the next response.
        return response

    @property
    def reply_to(self):
        """Return the dynamic address of our receiver."""
        return self.receiver.remote_source.address

    def on_message(self, event):
        """Called when we receive a message for our receiver."""
        self.response = event.message
        self.connection.container.yield_()  # Wake up the wait() loop to handle the message.