"""
NSQ base reader class.

This receives messages from nsqd and calls task methods to process that message

It handles the logic for backing off on retries and giving up on a message

ex.
    import nsq
    
    all_tasks = {"task1": task1, "task2": task2}
    r = nsq.Reader(all_tasks, lookupd_http_addresses=['127.0.0.1:4161'],
            topic="nsq_reader", channel="asdf", lookupd_poll_interval=15)
    nsq.run()
"""
import logging
import os
import ujson as json
import time
import signal
import socket
import functools
import urllib

import tornado.options
import tornado.ioloop
import tornado.httpclient

import BackoffTimer
import nsq
import async


tornado.options.define('heartbeat_file', type=str, default=None, help="path to a file to touch for heartbeats")

class RequeueWithoutBackoff(Exception):
    """exception for requeueing a message without incrementing backoff"""
    pass


class Reader(object):
    def __init__(self, all_tasks, topic, channel,
                nsqd_tcp_addresses=None, lookupd_http_addresses=None,
                max_tries=5, max_in_flight=1, requeue_delay=90, lookupd_poll_interval=120,
                preprocess_method=None, validate_method=None, async=False):
        """
        Reader provides a loop that calls each task provided by ``all_tasks`` up to ``max_tries``
        requeueing on any failures with increasing multiples of ``requeue_delay`` between subsequent
        tries of each message.
        
        ``preprocess_method`` defines an optional method that can alter the message data before
        other task functions are called.
        ``validate_method`` defines an optional method that returns a boolean as to weather or not
        this message should be processed.
        ``all_tasks``  defines the a mapping of tasks and functions that individually will be called
        with the message data.
        ``async`` determines whether handlers will do asynchronous processing. If set to True, handlers
        must accept a keyword argument called "finisher" that will be a callable used to signal message
        completion, with a boolean argument indicating success
        """
        assert isinstance(all_tasks, dict)
        for key, method in all_tasks.items():
            assert callable(method), "key %s must have a callable value" % key
        if preprocess_method:
            assert callable(preprocess_method)
        if validate_method:
            assert callable(validate_method)
        assert isinstance(topic, (str, unicode)) and len(topic) > 0
        assert isinstance(channel, (str, unicode)) and len(channel) > 0
        assert isinstance(max_in_flight, int) and 0 < max_in_flight < 2500
        
        if nsqd_tcp_addresses:
            if not isinstance(nsqd_tcp_addresses, (list, set, tuple)):
                assert isinstance(nsqd_tcp_addresses, (str, unicode))
                nsqd_tcp_addresses = [nsqd_tcp_addresses]
        else:
            nsqd_tcp_addresses = []
        
        if lookupd_http_addresses:
            if not isinstance(lookupd_http_addresses, (list, set, tuple)):
                assert isinstance(lookupd_http_addresses, (str, unicode))
                lookupd_http_addresses = [lookupd_http_addresses]
        else:
            lookupd_http_addresses = []
        
        assert nsqd_tcp_addresses or lookupd_http_addresses
        
        self.topic = topic
        self.channel = channel
        self.nsqd_tcp_addresses = nsqd_tcp_addresses
        self.lookupd_http_addresses = lookupd_http_addresses
        self.requeue_delay = int(requeue_delay * 1000)
        self.max_tries = max_tries
        self.max_in_flight = max_in_flight
        self.lookupd_poll_interval = lookupd_poll_interval
        self.async=async
        
        self.task_lookup = all_tasks
        self.preprocess_method = preprocess_method
        self.validate_method = validate_method
        
        self.backoff_timer = dict((k, BackoffTimer.BackoffTimer(0, 120)) for k in self.task_lookup.keys())
        
        self.hostname = socket.gethostname()
        self.short_hostname = self.hostname.split('.')[0]
        self.conns = {}
        self.http_client = tornado.httpclient.AsyncHTTPClient()
        
        logging.info("starting reader for topic '%s'..." % self.topic)
        
        for task in self.task_lookup: