class Channel:
    """
    A queue for one data type
    Uses BrightsideConsumer to be independent of the underlying implementation of the consumer i.e. rmq, redis, etc.
    It uses a buffer over the backing service queue, this allows us to insert control messages into the channel.
    We use control message to stop the consumption of messages from the channel
    """
    def __init__(self, name: str, consumer: BrightsideConsumer, pipeline: Queue) -> None:
        self._consumer = consumer
        self._name = ChannelName(name)
        self._queue = pipeline
        self._state = ChannelState.initialized
        self._cancel_heartbeat = None  # type: Event

    def __len__(self):
        return self._queue.qsize()

    def acknowledge(self, message: BrightsideMessage):
        self._consumer.acknowledge(message)

    def end(self) -> None:
        self._consumer.stop()
        self._state = ChannelState.stopped

    def end_heartbeat(self) -> None:
        if self._cancel_heartbeat is not None:
            self._cancel_heartbeat.set()

    @property
    def name(self) -> ChannelName:
        return self._name

    def receive(self, timeout: float) -> BrightsideMessage:
        if self._state is ChannelState.stopped:
            raise ChannelFailureException("Channel has been stopped, cannot resume listening")

        if self._state is ChannelState.initialized:
            self._state = ChannelState.started

        if not self._queue.empty():
            return self._queue.get(block=True, timeout=timeout)

        return self._consumer.receive(timeout=timeout)

    @property
    def state(self) -> ChannelState:
        return self._state

    def start_heartbeat(self):
        self. _cancel_heartbeat = self._consumer.run_heartbeat_continuously()

    def stop(self):
        self._queue.put(create_quit_message())
        self._state = ChannelState.stopping

    def requeue(self, message):
        self._consumer.requeue(message) class CommandProcessor:
    def __init__(self,
                 registry: Optional[Registry]=None,
                 message_mapper_registry: Optional[MessageMapperRegistry]=None,
                 message_store: Optional[BrightsideMessageStore]=None,
                 producer: Optional[BrightsideProducer]=None) -> None:
    def send(self, request: Request) -> None:
    def publish(self, request: Request) -> None:
    def post(self, request: Request) -> None: class Connection:
    """Contains the details required to connect to a RMQ broker: the amqp uri and the exchange"""
    def __init__(self, amqp_uri: str, exchange: str, exchange_type: str = "direct", is_durable: bool = False, connect_timeout: int = 30, heartbeat: int = 30) -> None:
        self._amqp_uri = amqp_uri
        self._exchange = exchange
        self._exchange_type = exchange_type
        self._is_durable = is_durable
        self._connect_timeout = connect_timeout
        self._heartbeat = heartbeat

    @property
    def amqp_uri(self) -> str:
        return self._amqp_uri

    @amqp_uri.setter
    def amqp_uri(self, value: str):
        self._amqp_uri = value

    @property
    def exchange(self) -> str:
        return self._exchange

    @exchange.setter
    def exchange(self, value: str):
        self._exchange = value

    @property
    def exchange_type(self) -> str:
        return self._exchange_type

    @exchange_type.setter
    def exchange_type(self, value: str):
        self._exchange_type = value

    @property
    def is_durable(self):
        return self._is_durable

    @is_durable.setter
    def is_durable(self, value):
        self._is_durable = value

    @property
    def connect_timeout(self) -> int:
        return self._connect_timeout
    
    @connect_timeout.setter
    def connect_timeout(self, value:int):
        self._connect_timeout = value

    @property
    def heartbeat(self) -> int:
        return self._heartbeat
    
    @heartbeat.setter
    def heartbeat(self, value:int):
        self._heartbeat = value class ConfigurationException(Exception):
    pass class MessagingException(Exception):
    pass def create_quit_message():
    body = BrightsideMessageBody(body="")
    header = BrightsideMessageHeader(uuid4(), topic="", message_type=BrightsideMessageType.MT_QUIT)
    return BrightsideMessage(header, body) class MessagePump:
    def __init__(self, command_processor: CommandProcessor,
                 channel: Channel,
                 mapper_func: Callable[[BrightsideMessage], Request],
                 timeout: int = None,
                 unacceptable_message_limit: int = None,
                 requeue_count: int = None) -> None:
        self._command_processor = command_processor
        self._channel = channel
        self._mapper_func = mapper_func
        self._timeout = timeout / 1000 if timeout else 0.5
        self._logger = logging.getLogger(__name__)
        self._unacceptable_message_limit = unacceptable_message_limit if unacceptable_message_limit else 500
        self._unacceptable_message_count = 0
        self._requeue_count = requeue_count

    def run(self, started_event: Event = None) -> None:

        # Can be used to signal that we have started to caller
        if started_event is not None:
            started_event.set()

        while True:

            if self._unacceptable_message_limit_reached():
                self._channel.end()
                break

            message = None
            try:
                self._logger.debug("MessagePump: Receiving messages from {} on thread # {}".format(
                    self._channel.name, current_thread().name))

                message = self._channel.receive(self._timeout)
            except ChannelFailureException:
                self._logger.warning("MessagePump: ChannelFailureException receiving messages from {} on thread # {}".format(
                    self._channel.name, current_thread().name), exc_info=1)
                continue
            except Exception:
                self._logger.warning("MessagePump: Exception receiving messages from {} on thread # {}".format(
                    self._channel.name, current_thread().name), exc_info=1)

            if message is None:
                raise ChannelFailureException("Could not receive message. Note that should return BrightsideMessageType.none from an empty queeu")
            elif message.header.message_type == BrightsideMessageType.MT_NONE:
                time.sleep(self._timeout)
                continue
            elif message.header.message_type == BrightsideMessageType.MT_QUIT:
                self._logger.debug("MessagePump: Quit receiving messages from {} on thread # {}".format(
                    self._channel.name, current_thread().name))
                self._channel.end()
                break
            elif message.header.message_type == BrightsideMessageType.MT_UNACCEPTABLE:
                self._logger.debug("MessagePump: Failed to parse a message from the incoming message with id {} from {} on thread # {} ".format(
                    message.id, self._channel.name, current_thread().name))
                self._acknowledge_message(message)
                self._increment_unacceptable_message_count()
                continue

            with (heartbeat(self._channel)):
                try:
                    # Serviceable message
                    request = self._translate_message(message)
                    self._dispatch_message(message.header, request)

                except DeferMessageException:
                    self._requeue_message(message)
                    continue
                except ConfigurationException:
                    raise
                except Exception as ex:
                    self._logger.error("MessagePump: Failed to dispatch the message with id {} from {} on thread # {} due to {}".format(
                        message.id, self._channel.name, current_thread().name, ex))

                self._acknowledge_message(message)

        self._logger.debug("MessagePump: Finished running message loop, no longer receiving messages from {} on thread # {}".format(
            self._channel.name, current_thread().name))

        self._channel.end()

    def _acknowledge_message(self, message: BrightsideMessage) -> None:
        self._logger.debug("MessagePump: Acknowledge message {} from {} on thread # {}".format(
            message.id, self._channel.name, current_thread().name))
        self._channel.acknowledge(message)

    def _discard_requeued_messages_enabled(self):
        return self._requeue_count is not None

    def _dispatch_message(self, message_header: BrightsideMessageHeader, request: Request) -> None:
        if message_header.message_type == BrightsideMessageType.MT_COMMAND:
            self._command_processor.send(request)
        elif message_header.message_type == BrightsideMessageType.MT_EVENT:
            self._command_processor.publish(request)

    def _increment_unacceptable_message_count(self) -> int:
        self._unacceptable_message_count += 1
        return self._unacceptable_message_count

    def _requeue_message(self, message: BrightsideMessage) -> None:
        message.increment_handled_count()

        if self._discard_requeued_messages_enabled():
            if message.handled_count_reached(self._requeue_count):
                self._logger.error("MessagePump: Have tried {} times to handle this message {} dropping message \n. Message Body {} ".format(
                    self._requeue_count, message.id, message.body.value))
                self._channel.acknowledge(message)
                return

        self._logger.debug("MessagePump: Re-queueing message {} from {}".format(
                message.id, self._channel.name))
        self._channel.requeue(message)

    def _translate_message(self, message: BrightsideMessage)-> Request:
        if self._mapper_func is None:
            raise ConfigurationException("Missing Mapper Function for message topic {}".format(message.header.topic))
        return self._mapper_func(message)

    def _unacceptable_message_limit_reached(self) -> bool:
        return self._unacceptable_message_count >= self._unacceptable_message_limit class BrightsideConsumerConfiguration:
    """
    A class to encapsulate the configuration required to create a Brightside Consumer
    Required because we need to pass the parameters to a performer that runs the message pump on another thread
    Use is_long_running_handler when you expect a handler to take more than about 10s to execute, this will instruct
    the consumer to spin up a separate thread to send a heartbeat over the connection. Due to the GIL this won't help
    you that much if your handler is CPU bound though.
    """
    def __init__(self, pipeline: Queue, queue_name: str, routing_key: str, prefetch_count: int=1,
                 is_durable: bool=False, is_ha: bool=False, is_long_running_handler: bool=False):
        self._pipeline = pipeline
        self._queue_name = queue_name
        self._routing_key = routing_key
        self._prefetch_count = prefetch_count
        self._is_durable = is_durable
        self._is_ha = is_ha
        self._is_long_running = is_long_running_handler

    @property
    def pipeline(self):
        return self._pipeline

    @property
    def queue_name(self) -> str:
        return self._queue_name

    @property
    def routing_key(self) -> str:
        return self._routing_key

    @property
    def prefetch_count(self) -> int:
        return self._prefetch_count

    @property
    def is_durable(self) -> bool:
        return self._is_durable

    @property
    def is_ha(self) -> bool:
        return self._is_ha

    @property
    def is_long_runing_handler(self) -> bool:
        return self._is_long_running class BrightsideConsumer(metaclass=ABCMeta):
    """The comoonent that receives messages from a broker. Usually abstracts a queue for subscribing to a topic on the
    broker i.e. a dynamic recepient list.
    """

    @abstractmethod
    def acknowledge(self, message: BrightsideMessage):
        pass

    @abstractmethod
    def has_acknowledged(self, message):
        pass

    @abstractmethod
    def purge(self):
        pass

    @abstractmethod
    def receive(self, timeout: float) -> BrightsideMessage:
        pass

    @abstractmethod
    def requeue(self, message) -> None:
        pass

    @abstractmethod
    def run_heartbeat_continuously(self) -> Event:
        pass

    @abstractmethod
    def stop(self):
        pass class BrightsideMessage:
    """The representation of an on the wire message in Brighter. It abstracts the message typeof the underlying
    implementation and thus acts as a anti-corruption layer between us and an implementation specific message
    type
    """
    def __init__(self, message_header: BrightsideMessageHeader, message_body: BrightsideMessageBody) -> None:
        self._message_header = message_header
        self._message_body = message_body

    @property
    def header(self) -> BrightsideMessageHeader:
        return self._message_header

    @property
    def body(self) -> BrightsideMessageBody:
        return self._message_body

    def handled_count_reached(self, requeue_count: int) -> bool:
        return self._message_header.handled_count >= requeue_count

    @property
    def id(self) -> UUID:
        return self._message_header.id

    def increment_handled_count(self):
        self._message_header.increment_handled_count()
import logging
import time
from enum import Enum
from multiprocessing import Event, Process
from threading import Thread
from typing import Callable, Dict
from brightside.channels import Channel
from brightside.command_processor import CommandProcessor, Request
from brightside.connection import Connection
from brightside.exceptions import ConfigurationException, MessagingException
from brightside.message_factory import create_quit_message
from brightside.message_pump import MessagePump
from brightside.messaging import BrightsideConsumerConfiguration, BrightsideConsumer, BrightsideMessage
"""
File             : dispatch.py
Author           : ian
Created          : 04-21-2017

Last Modified By : ian
Last Modified On : 04-21-2017
***********************************************************************
The MIT License (MIT)
Copyright © 2017 Ian Cooper <ian_hammond_cooper@yahoo.co.uk>

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the “Software”), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
***********************************************************************
"""



class Performer:
    def __init__(self,
                 channel_name: str,
                 connection: Connection,
                 consumer_configuration: BrightsideConsumerConfiguration,
                 consumer_factory: Callable[[Connection, BrightsideConsumerConfiguration, logging.Logger], BrightsideConsumer],
                 command_processor_factory: Callable[[str], CommandProcessor],
                 mapper_func: Callable[[BrightsideMessage], Request],
                 logger: logging.Logger=None
                 ) -> None:
        """
        Each Performer abstracts a process running a message pump.
        That process is forked from the parent, as we cannot guarantee a message pump is only I/O bound and thus will
        not scale because of the GIL.
        The Performer is how the supervisor (the dispatcher) tracks the workers it has created
        The Performer needs:
        :param channel_name: The name of the channel we want to create a sub-process for
        :param connection: The connection to the broker
        :param consumer_factory: We need a user supplied callback to provide us an instance of the concumer for
            the broker we are using. Arame? Something else?
        :param command_processor_factory: We need a user supplied callback to create a commandprocessor with
            subscribers, policies, outgoing tasks queues etc.
        :param mapper_func: We need a user supplied callback to map on the wire messages to requests
        """
        # TODO: The paramater needs to be a connection, not an AramaConnection as we can't decide to create an Arame Consumer
        # here. Where do we make that choice?

        self._channel_name = channel_name
        self._connection = connection
        self._consumer_configuration = consumer_configuration
        self._consumer_factory = consumer_factory
        self._command_processor_factory = command_processor_factory
        self._mapper_func = mapper_func
        self._logger = logger or logging.getLogger(__name__)

    def stop(self) -> None:
        self._consumer_configuration.pipeline.put(create_quit_message())

    def run(self, started_event: Event) -> Process:

        p = Process(target=_sub_process_main, args=(
            started_event,
            self._channel_name,
            self._connection,
            self._consumer_configuration,
            self._consumer_factory,
            self._command_processor_factory,
            self._mapper_func))

        self._logger.debug("Starting worker process for channel: %s on exchange %s on server %s",
                           self._channel_name, self._connection.exchange, self._connection.amqp_uri)

        p.start()

        started_event.wait(timeout=1)

        return p


def _sub_process_main(started_event: Event,
                      channel_name: str,
                      connection: Connection,
                      consumer_configuration: BrightsideConsumerConfiguration,
                      consumer_factory: Callable[[Connection, BrightsideConsumerConfiguration, logging.Logger], BrightsideConsumer],
                      command_processor_factory: Callable[[str], CommandProcessor],
                      mapper_func: Callable[[BrightsideMessage], Request]) -> None:
    """
    This is the main method for the sub=process, everything we need to create the message pump and
    channel it needs to be passed in as parameters that can be pickled as when we run they will be serialized
    into this process. The data should be value types, not reference types as we will receive a copy of the original.
    Inter-process communication is signalled by the event - to indicate startup - and the pipeline to facilitate a
    sentinel or stop message
    :param started_event: Used by the sub-process to signal that it is ready
    :param channel_name: The name we want to give the channel to the broker for identification
    :param connection: The 'broker' connection
    :param consumer_configuration: How to configure our consumer of messages from the channel
    :param consumer_factory: Callback to create the consumer. User code as we don't know what consumer library they
        want to use. Arame? Something else?
    :param command_processor_factory: Callback to  register subscribers, policies, and task queues then build command
        processor. User code that provides us with their requests and handlers
    :param mapper_func: We need to map between messages on the wire and our handlers
    :return:
    """

    logger = logging.getLogger(__name__)
    consumer = consumer_factory(connection, consumer_configuration, logger)

    channel = Channel(name=channel_name, consumer=consumer, pipeline=consumer_configuration.pipeline)