import multiprocessing


def consumer_process_factory(kafka_bootstrap_servers):
    """
    Use this "factory as a fixture" to create multiprocessing.Process running
    a Kafka consumer polling loop in a test function.

    Parameters
    ----------
    kafka_bootstrap_servers: pytest fixture (str)
        comma-separated list of host:port

    Returns
    -------
    _consumer_process_factory :
            function(consumer_factory, topics, group_id, consumer_config, document_queue, **kwargs)
        factory function returning a multiprocessing.Process that will run a Kafka
        consumer polling loop

    """

    def _consumer_process_factory(
        consumer_factory, topics, group_id, consumer_config, document_queue, **kwargs
    ):
        """
        Parameters
        ----------
        consumer_factory : function(topics, group_id, consumer_config, **kwargs)
            a factory function (or callable) returning a BlueskyConsumer-like object
        topics : list of str, required
            the underlying Kafka consumer will subscribe to the specified topics
        group_id : str, optional
            the underlying Kafka consumer will have the specified group_id, "pytest" by default
        consumer_config : dict, optional
            the underlying Kafka consumer will be created with the specified configuration parameters
        document_queue : multiprocessing.Queue
            the underlying Kafka consumer will place documents it receives in this queue
        kwargs
            kwargs will be passed to the consumer_factory to be used in constructing the
            underlying Kafka consumer

        Returns
        -------
        document_queue: multiprocessing.Queue
            this queue will contain bluesky (name, document) pairs that were delivered
            to the underlying Kafka consumer
        """
        if consumer_config is None:
            consumer_config = {
                # it is important to set a short time interval
                # for automatic commits or the Kafka broker may
                # not be notified by the consumer that messages
                # were received before the test ends; the result
                # is that the Kafka broker will try to re-deliver
                # those messages to the next consumer that subscribes
                # to the same topic(s)
                "auto.commit.interval.ms": 100,
            }

        # this function will run in the external process created below
        def start_consumer_with_queue(document_queue_):
            logger = multiprocessing.get_logger()
            logger.warning("constructing consumer process with inter-process queue")

            def put_document_in_queue(consumer, topic, name, doc):
                logger.warning("BlueskyConsumer putting %s in queue", name)
                document_queue_.put((name, doc))

            # it is important the BlueskyConsumer be
            # constructed in the external process
            bluesky_consumer_ = consumer_factory(
                topics=topics,
                bootstrap_servers=kafka_bootstrap_servers,
                group_id=group_id,
                consumer_config=consumer_config,
                process_document=put_document_in_queue,
                **kwargs,
            )
            # consume messages published by a Kafka broker
            bluesky_consumer_.start()

        # create an external process for the bluesky_kafka.BlueskyConsumer polling loop
        # but do not start it, the client of this function will start the process
        consumer_process = multiprocessing.Process(
            target=start_consumer_with_queue, args=(document_queue,), daemon=True,
        )

        return consumer_process

    return _consumer_process_factory