# vim: set fileencoding=utf-8 :

###################################################################################
#                                                                                 #
# Copyright (c) 2019 Idiap Research Institute, http://www.idiap.ch/               #
# Contact: beat.support@idiap.ch                                                  #
#                                                                                 #
# Redistribution and use in source and binary forms, with or without              #
# modification, are permitted provided that the following conditions are met:     #
#                                                                                 #
# 1. Redistributions of source code must retain the above copyright notice, this  #
# list of conditions and the following disclaimer.                                #
#                                                                                 #
# 2. Redistributions in binary form must reproduce the above copyright notice,    #
# this list of conditions and the following disclaimer in the documentation       #
# and/or other materials provided with the distribution.                          #
#                                                                                 #
# 3. Neither the name of the copyright holder nor the names of its contributors   #
# may be used to endorse or promote products derived from this software without   #
# specific prior written permission.                                              #
#                                                                                 #
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND #
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED   #
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE          #
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE    #
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL      #
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR      #
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER      #
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,   #
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE   #
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.            #
#                                                                                 #
###################################################################################


"""BEAT Computation broker

Inspired by the Majordomo Protocol Broker
"""
import logging
import signal
import time
from binascii import hexlify

import zmq

from . import BCP
from .zhelpers import dump

logger = logging.getLogger(__name__)


class Service(object):
    """a single Service"""

    name = None  # Service name
    requests = None  # List of client requests
    waiting = None  # List of waiting workers

    def __init__(self, name):
        self.name = name
        self.requests = []
        self.waiting = []


class Worker(object):
    """a Worker, idle or active"""

    identity = None  # hex Identity of worker
    address = None  # Address to route to
    service = None  # Owning service, if known
    expiry = None  # expires at this point, unless heartbeat

    def __init__(self, identity, address, lifetime):
        self.identity = identity
        self.address = address
        self.expiry = time.time() + 1e-3 * lifetime

    def __str__(self):
        return "{}:{}:{}".format(self.identity, self.address, self.expiry)


class BeatComputationBroker(object):
    """
    Beat Computation Protocol broker
    """

    HEARTBEAT_LIVENESS = 3  # 3-5 is reasonable
    HEARTBEAT_INTERVAL = 2500  # msecs
    HEARTBEAT_EXPIRY = HEARTBEAT_INTERVAL * HEARTBEAT_LIVENESS

    def __init__(self, verbose=False):
        """Initialize broker state."""

        self.verbose = verbose
        self.continue_ = True
        self.services = {}
        self.workers = {}
        self.waiting = []
        self.heartbeat_at = time.time() + 1e-3 * self.HEARTBEAT_INTERVAL
        self.ctx = zmq.Context()
        self.socket = self.ctx.socket(zmq.ROUTER)
        self.socket.linger = 0
        self.poller = zmq.Poller()
        self.poller.register(self.socket, zmq.POLLIN)
        self.callbacks = {}

        signal.signal(signal.SIGTERM, self.__signal_handler)
        signal.signal(signal.SIGINT, self.__signal_handler)

    def __signal_handler(self, signum, frame):
        self.continue_ = False

    def set_worker_callbacks(self, on_connection, on_disconnection):
        self.callbacks = {
            "on_connection": on_connection,
            "on_disconnection": on_disconnection,
        }

    def mediate(self):
        """Main broker work happens here"""

        while self.continue_:

            items = self.poller.poll(self.HEARTBEAT_INTERVAL)

            if not self.continue_:
                # Received a signal
                break

            if items:
                msg = self.socket.recv_multipart()
                if self.verbose:
                    logger.info("I: received message:")
                    dump(msg)

                sender = msg.pop(0)
                empty = msg.pop(0)
                assert empty == b""  # nosec
                header = msg.pop(0)

                if BCP.BCPC_CLIENT == header:
                    self.process_client(sender, msg)
                elif BCP.BCPW_WORKER == header:
                    self.process_worker(sender, msg)
                else:
                    logger.error("E: invalid message:")
                    dump(msg)

            self.purge_workers()
            self.send_heartbeats()

    def destroy(self):
        """Disconnect all workers, destroy context."""

        while self.workers:
            self.delete_worker(self.workers.values()[0], True)
        self.ctx.destroy(0)

    def process_client(self, sender, msg):
        """Process a request coming from a client."""
        # Service name + body

        assert len(msg) >= 2  # nosec
        service = msg.pop(0)

        # Set reply return address to client sender
        msg = [sender, b""] + msg
        self.dispatch(self.require_service(service), msg)

    def process_worker(self, sender, msg):
        """Process message sent to us by a worker."""

        # At least, command
        assert len(msg) >= 1  # nosec

        command = msg.pop(0)

        worker_ready = hexlify(sender) in self.workers

        worker = self.require_worker(sender)

        if BCP.BCPW_READY == command:
            # At least, a service name
            assert len(msg) >= 1  # nosec
            service = msg.pop(0)
            # Not first command in session or Reserved service name
            if worker_ready:
                self.delete_worker(worker, True)
            else:
                # Attach worker to service and mark as idle
                worker.service = self.require_service(service)
                self.worker_waiting(worker)

                on_connection = self.callbacks.get("on_connection", None)
                if on_connection:
                    on_connection(worker.service.name)

        elif BCP.BCPW_REPLY == command:
            if worker_ready:
                # Remove & save client return envelope and insert the
                # protocol header and service name, then rewrap envelope.
                client = msg.pop(0)
                msg.pop(0)  # empty
                msg = [client, b"", BCP.BCPC_CLIENT, worker.service.name] + msg
                self.socket.send_multipart(msg)
                self.worker_waiting(worker)
            else:
                self.delete_worker(worker, True)

        elif BCP.BCPW_HEARTBEAT == command:
            if worker_ready:
                worker.expiry = time.time() + 1e-3 * self.HEARTBEAT_EXPIRY
            else:
                self.delete_worker(worker, True)

        elif BCP.BCPW_DISCONNECT == command:
            self.delete_worker(worker, False)
        else:
            logger.error("E: invalid message:")
            dump(msg)

    def delete_worker(self, worker, disconnect):
        """Deletes worker from all data structures, and deletes worker."""

        assert worker is not None  # nosec
        if disconnect:
            self.send_to_worker(worker, BCP.BCPW_DISCONNECT, None, None)

        if worker.service is not None:
            if worker in worker.service.waiting:
                worker.service.waiting.remove(worker)

            on_disconnection = self.callbacks.get("on_disconnection", None)
            if on_disconnection:
                on_disconnection(worker.service.name)

        if worker.identity in self.workers:
            self.workers.pop(worker.identity)

        if worker in self.waiting:
            self.waiting.pop(self.waiting.index(worker))

    def require_worker(self, address):
        """Finds the worker (creates if necessary)."""

        assert address is not None  # nosec
        identity = hexlify(address)
        worker = self.workers.get(identity)
        if worker is None:
            worker = Worker(identity, address, self.HEARTBEAT_EXPIRY)
            self.workers[identity] = worker
            if self.verbose:
                logger.info("I: registering new worker: %s", identity)

        return worker

    def require_service(self, name):
        """Locates the service (creates if necessary)."""

        assert name is not None  # nosec
        service = self.services.get(name)
        if service is None:
            service = Service(name)
            self.services[name] = service

        return service

    def bind(self, endpoint):
        """Bind broker to endpoint, can call this multiple times.

        We use a single socket for both clients and workers.
        """

        self.socket.bind(endpoint)
        logger.info("I: BCP broker/0.0.1 is active at %s", endpoint)

    def send_heartbeats(self):
        """Send heartbeats to idle workers if it's time"""

        if time.time() > self.heartbeat_at:
            for worker in self.waiting:
                self.send_to_worker(worker, BCP.BCPW_HEARTBEAT, None, None)

            self.heartbeat_at = time.time() + 1e-3 * self.HEARTBEAT_INTERVAL

    def purge_workers(self):
        """Look for & kill expired workers."""

        for item in self.waiting:
            if item.expiry < time.time():
                logger.info("I: deleting expired worker: %s", item.identity)
                self.delete_worker(item, False)

    def worker_waiting(self, worker):
        """This worker is now waiting for work."""

        # Queue to broker and service waiting lists

        if worker not in self.waiting:
            self.waiting.append(worker)
        if worker not in worker.service.waiting:
            worker.service.waiting.append(worker)

        worker.expiry = time.time() + 1e-3 * self.HEARTBEAT_EXPIRY
        self.dispatch(worker.service, None)

    def dispatch(self, service, msg):
        """Dispatch requests to waiting workers as possible"""

        assert service is not None  # nosec

        if msg is not None:  # Queue message if any
            service.requests.append(msg)

        self.purge_workers()

        while service.waiting and service.requests:
            msg = service.requests.pop(0)
            worker = service.waiting.pop(0)
            self.waiting.remove(worker)
            self.send_to_worker(worker, BCP.BCPW_REQUEST, None, msg)

    def send_to_worker(self, worker, command, option, msg=None):
        """Send message to worker.

        If message is provided, sends that message.
        """

        if msg is None:
            msg = []
        elif not isinstance(msg, list):
            msg = [msg]

        # Stack routing and protocol envelopes to start of message
        # and routing envelope
        if option is not None:
            msg = [option] + msg
        msg = [worker.address, b"", BCP.BCPW_WORKER, command] + msg

        if self.verbose:
            logger.info("I: sending %r to worker", command)
            dump(msg)

        self.socket.send_multipart(msg)
