# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2012 Rackspace
# All Rights Reserved.
#
#    Licensed 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.

import logging

import kombu

from tempo import actions
from tempo import config
from tempo import db
from tempo import notifier
from tempo import queue as tempo_queue
from tempo.openstack.common import cfg
from tempo.openstack.common import exception as common_exception

CFG = config.CFG

logger = logging.getLogger('tempo.worker')

worker_opts = [
    cfg.BoolOpt('daemonized',
                default=False,
                help='Run worker as a daemon'),
    cfg.StrOpt('publisher_id',
               default='host',
               help='Where the notification came from')
]

worker_group = cfg.OptGroup(name='worker', title='Worker options')
CFG.register_group(worker_group)
CFG.register_opts(worker_opts, group=worker_group)


def _perform_task(task):
    def _notify(event_type, exception=None):
        payload = {'task_uuid': task_uuid}
        if exception is not None:
            payload['exception'] = exception

        publisher_id = CFG.worker.publisher_id
        priority = notifier.DEBUG
        notifier.notify(publisher_id, event_type, priority, payload)

    action = task.action
    task_uuid = task.uuid
    try:
        func = getattr(actions, action)
    except AttributeError:
        logger.error("unrecognized action '%(action)s' for task task"
                     " '%(task_uuid)s'" % locals())
        return

    logger.debug("task '%(task_uuid)s' started: '%(action)s'" % locals())

    _notify('Started Task')

    try:
        func(task)
    except Exception as e:
        logger.error("task '%(task_uuid)s' errored: %(e)s" % locals())
        _notify('Errored Task', exception=e)
    else:
        logger.debug("task '%(task_uuid)s' finished: returned successfully" %
                     locals())
        _notify('Finished Task')


def _process_message(body, message):
    message.ack()

    task_uuid = body['task_uuid']

    try:
        task = db.task_get(task_uuid)
    except common_exception.NotFound:
        logger.error("Task '%(task_uuid)s' not found" % locals())
        return

    _perform_task(task)


def _consume_messages(exchange, queue, key):
    kombu_xchg = kombu.Exchange(exchange, 'direct', durable=True)
    kombu_queue = kombu.Queue(queue, exchange=kombu_xchg, key=key)

    connection = tempo_queue.get_connection()

    consumer = kombu.Consumer(connection.channel(), kombu_queue)
    consumer.register_callback(_process_message)
    consumer.consume()

    while True:
        connection.drain_events()


def consume_messages(exchange, queue, key):
    if CFG.worker.daemonized:
        # TODO(mdietz): there's a cleaner way to do this, but this works well
        # as a way of backgrounding the server for now
        import daemon
        with daemon.DaemonContext():
            _consume_messages(exchange, queue, key)
    else:
        _consume_messages(exchange, queue, key)

[((970, 1003), 'logging.getLogger', 'logging.getLogger', (['"""tempo.worker"""'], {}), "('tempo.worker')\n", (987, 1003), False, 'import logging\n'), ((1266, 1317), 'tempo.openstack.common.cfg.OptGroup', 'cfg.OptGroup', ([], {'name': '"""worker"""', 'title': '"""Worker options"""'}), "(name='worker', title='Worker options')\n", (1278, 1317), False, 'from tempo.openstack.common import cfg\n'), ((1025, 1096), 'tempo.openstack.common.cfg.BoolOpt', 'cfg.BoolOpt', (['"""daemonized"""'], {'default': '(False)', 'help': '"""Run worker as a daemon"""'}), "('daemonized', default=False, help='Run worker as a daemon')\n", (1036, 1096), False, 'from tempo.openstack.common import cfg\n'), ((1134, 1222), 'tempo.openstack.common.cfg.StrOpt', 'cfg.StrOpt', (['"""publisher_id"""'], {'default': '"""host"""', 'help': '"""Where the notification came from"""'}), "('publisher_id', default='host', help=\n    'Where the notification came from')\n", (1144, 1222), False, 'from tempo.openstack.common import cfg\n'), ((2788, 2836), 'kombu.Exchange', 'kombu.Exchange', (['exchange', '"""direct"""'], {'durable': '(True)'}), "(exchange, 'direct', durable=True)\n", (2802, 2836), False, 'import kombu\n'), ((2855, 2903), 'kombu.Queue', 'kombu.Queue', (['queue'], {'exchange': 'kombu_xchg', 'key': 'key'}), '(queue, exchange=kombu_xchg, key=key)\n', (2866, 2903), False, 'import kombu\n'), ((2922, 2950), 'tempo.queue.get_connection', 'tempo_queue.get_connection', ([], {}), '()\n', (2948, 2950), True, 'from tempo import queue as tempo_queue\n'), ((1686, 1746), 'tempo.notifier.notify', 'notifier.notify', (['publisher_id', 'event_type', 'priority', 'payload'], {}), '(publisher_id, event_type, priority, payload)\n', (1701, 1746), False, 'from tempo import notifier\n'), ((2557, 2579), 'tempo.db.task_get', 'db.task_get', (['task_uuid'], {}), '(task_uuid)\n', (2568, 2579), False, 'from tempo import db\n'), ((3384, 3406), 'daemon.DaemonContext', 'daemon.DaemonContext', ([], {}), '()\n', (3404, 3406), False, 'import daemon\n')]