"""Implementation of the service."""

import sys

import click

from ingestion.kafka import connect, ConnectionInfo, Consumer
from ingestion.importer import import_from_service, ServiceImportError


class Application:

    """This is a class."""

    _consumer = None

    def __init__(self, *, host, port, topic, callback, group=None):
        """Initialize the class."""
        self.info = ConnectionInfo(host, port, topic, group)
        self.callback = callback

    def _initialize(self):
        client = connect(host=self.info.host, port=self.info.port)

        self._consumer = Consumer(
            client=client, topic=self.info.topic, group=self.info.group)

    def run_forever(self):
        """Run the class."""
        if not self._consumer:
            self._initialize()

        messages = self._consumer.read()
        while True:
            try:
                message = next(messages)
            except KeyboardInterrupt:
                break
            except Exception:
                continue

            self.callback(message)


@click.command(context_settings={'help_option_names': ['-h', '--help']})
@click.argument('service')
def run(service):
    """Run an ingestion service."""
    try:
        settings = import_from_service(service, 'settings')
    except ServiceImportError as e:
        raise click.BadOptionUsage('service', str(e))
    try:
        process = import_from_service(service, 'process')
    except ServiceImportError as e:
        raise click.BadOptionUsage('service', str(e))
    app = Application(
        host=settings.KAFKA_BROKER_HOST,
        port=settings.KAFKA_BROKER_PORT,
        topic=settings.KAFKA_TOPIC_INBOUND,
        group=settings.KAFKA_GROUP_NAME,
        callback=process.run,
    )
    app.run_forever()


if __name__ == '__main__':
    sys.exit(run())

@@ -223,116 +223,203 @@
  %22%22%22
-This is a class.%22%22%22%0A%0A    _consumer = None%0A%0A    def __init__(self, *, host, port, topic, callback, group=None
+A service application.%0A%0A    A service application will spin up its own Kafka consumer and pass%0A    each incoming message to a callback.%0A    %22%22%22%0A%0A    def __init__(self, name, settings, *, callback
 ):%0A 
@@ -441,35 +441,33 @@
 alize the class.
-%22%22%22
+%0A
 %0A        self.in
@@ -463,93 +463,780 @@
     
-self.info = ConnectionInfo(host, port, topic, group)%0A        self.callback = callback
+Args:%0A            name (str): The name of the application.%0A            settings (object): An object with attributed-based settings.%0A            callback (callable): A callable object that takes two%0A              arguments, an instance of this class and a dict%0A              representing an incoming message.%0A        %22%22%22%0A        self.name = name%0A        self.settings = settings%0A        self.callback = callback%0A%0A        # KafkaClient connections are eager. Don't instantiate an%0A        # instance until we're ready to use it.%0A        self._consumer = None%0A        self._consumer_info = ConnectionInfo(%0A            settings.KAFKA_BROKER_HOST,%0A            settings.KAFKA_BROKER_PORT,%0A            settings.KAFKA_TOPIC_INBOUND,%0A            settings.KAFKA_GROUP_NAME,%0A        )
 %0A%0A  
@@ -1289,18 +1289,41 @@
 ect(
-host=self.
+%0A            host=self._consumer_
 info
@@ -1335,24 +1335,34 @@
 , port=self.
+_consumer_
 info.port)%0A%0A
@@ -1422,16 +1422,28 @@
 =client,
+%0A           
  topic=s
@@ -1442,24 +1442,34 @@
  topic=self.
+_consumer_
 info.topic, 
@@ -1471,30 +1471,62 @@
 pic,
- group=self.info.group
+%0A            group=self._consumer_info.group,%0A        
 )%0A%0A 
@@ -1904,16 +1904,22 @@
 allback(
+self, 
 message)
@@ -2426,80 +2426,20 @@
     
-host=settings.KAFKA_BROKER_HOST,%0A        port=settings.KAFKA_BROKER_PORT
+name=service
 ,%0A  
@@ -2448,22 +2448,16 @@
     
-topic=
 settings
 .KAF
@@ -2456,69 +2456,17 @@
 ings
-.KAFKA_TOPIC_INBOUND,%0A        group=settings.KAFKA_GROUP_NAME
+=settings
 ,%0A  

Restructure the Application class