7 files changed, 186 insertions, 104 deletions

diff --git a/docs/settings_file.rst b/docs/settings_file.rst
index 04fe451..e9f4b04 100644
--- a/docs/settings_file.rst
+++ b/docs/settings_file.rst
@@ -19,6 +19,13 @@ Default: False
 
 Enable most verbose level of debug statements
 
+hide_heartbeat_logs
+===================
+Default: True
+
+This hides heart beat messages from the logs. Disabling this will result in very
+noisy log output.
+
 max_sockets
 ===========
 Default: 1024
@@ -91,15 +98,55 @@ Default: ''
 
 Password to use when connecting to redis
 
+redis_client_class
+==================
+Default: ``redis.StrictRedis``
+
+The class to use as the redis client. This can be overridden if you want to use
+a different module to connect to redis. For example
+``rediscluster.StrictRedisCluster``. Note: You make get errors if you don't use
+a strict mode class.
+
+redis_client_class_kwargs
+=========================
+Default: {}
+
+This is a JSON hash map of keyword arguments to pass to the Python class
+constructor. This is useful for using ``redis-cluster-py`` on AWS Elasticache.
+When using Elasticache this value should be set to
+``{"skip_full_coverage_check": true}`` to prevent startup errors.
+
+redis_startup_error_hard_kill
+=============================
+Default: True
+
+If there is an error connecting to the Redis server for persistent schedule
+storage on startup then kill the app. This is useful if you want to prevent
+accidentally accepting schedules that can't be saved to a persistent store. If
+you would like to use redis you will need to ``pip install redis`` or
+``pip install redis-py-cluster`` and define the necessary options.
+
 ***********
 Job Manager
 ***********
 
+default_queue_name
+==================
+Default: default
+
+This is the default queue a job manager will listen on if nothing is specified.
+
+default_queue_weight
+====================
+Default: 10
+
+This is the default weight for the default queue is it is not explicitly set.
+
 concurrent_jobs
 ===============
 Default: 4
 
-This is the number of concurrent jobs the indiviudal job manager should execute
+This is the number of concurrent jobs the individual job manager should execute
 at a time. If you are using the multiprocess or threading model this number
 becomes important as you will want to control the load on your server. If the
 load equals the number of cores on the server, processes will begin waiting for
@@ -115,7 +162,7 @@ queues
 ======
 Default: [[10, "default"]]
 
-Comma seperated list of queues to process jobs for with their weights. This list
+Comma separated list of queues to process jobs with their weights. This list
 must be valid JSON otherwise an error will be thrown.
 Example: ``queues=[[10, "data_process"], [15, "email"]]``.  With these
 weights and the ``CONCURRENT_JOBS`` setting, you should be able to tune managers
@@ -131,14 +178,29 @@ number until the large box is ready to accept another q1 job.
    default queue so that anything that is not explicitly assigned will still be
    run.
 
-setup_callabe/setup_path
-========================
+setup_callable/setup_path
+=========================
 Default: '' (Signifies no task will be attempted)
 
 Strings containing path and callable to be run when a worker is spawned
 if applicable to that type of worker.  Currently the only supported worker is a
 MultiProcessWorker, and is useful for pulling any global state into memory.
 
+job_entry_func
+==============
+Default: '' (Signifies no function will be executed)
+
+The function to execute before **every** job a worker thread executes. For
+example: cleaning up stale database connections. (Django's
+``django.db.connections[].close_if_unusable_or_obsolete()``)
+
+job_exit_func
+=============
+Default: '' (Signifies no function will be executed)
+
+The function to execute **after** every job a worker thread executes. For
+example: closing any database handles that were left open.
+
 max_job_count
 =============
 Default: 1024
diff --git a/etc/eventmq.conf-dist b/etc/eventmq.conf-dist
index 96dbad4..7882e0c 100644
--- a/etc/eventmq.conf-dist
+++ b/etc/eventmq.conf-dist
@@ -19,6 +19,7 @@ scheduler_addr=tcp://eventmq:47291
 [router]
 
 [scheduler]
+redis_client_class = redis.StrictRedis
 
 [jobmanager]
 worker_addr=tcp://127.0.0.1:47290
diff --git a/eventmq/__init__.py b/eventmq/__init__.py
index 4143854..018a540 100644
--- a/eventmq/__init__.py
+++ b/eventmq/__init__.py
@@ -1,5 +1,5 @@
 __author__ = 'EventMQ Contributors'
-__version__ = '0.3.9'
+__version__ = '0.3.10-rc1'
 
 PROTOCOL_VERSION = 'eMQP/1.0'
 
diff --git a/eventmq/client/messages.py b/eventmq/client/messages.py
index 9e49ea8..c178c30 100644
--- a/eventmq/client/messages.py
+++ b/eventmq/client/messages.py
@@ -226,14 +226,14 @@ def send_request(socket, message, reply_requested=False, guarantee=False,
         }
     }
     Args:
-        socket (socket): Socket to use when sending `message`
+        socket: Socket (Sender or Receiver) to use when sending `message`
         message: message to send to `socket`
         reply_requested (bool): request the return value of func as a reply
         guarantee (bool): (Give your best effort) to guarantee that func is
             executed. Exceptions and things will be logged.
         retry_count (int): How many times should be retried when encountering
             an Exception or some other failure before giving up. (default: 0
-            or immediatly fail)
+            or immediately fail)
         timeout (int): How many seconds should we wait before killing the job
             default: 0 which means infinite timeout
         queue (str): Name of queue to use when executing the job. Default: is
diff --git a/eventmq/conf.py b/eventmq/conf.py
index f15a1ef..8634ebb 100644
--- a/eventmq/conf.py
+++ b/eventmq/conf.py
@@ -90,6 +90,7 @@ RQ_DB = 0
 RQ_PASSWORD = ''
 REDIS_CLIENT_CLASS = 'redis.StrictRedis'
 REDIS_CLIENT_CLASS_KWARGS = {}
+REDIS_STARTUP_ERROR_HARD_KILL = True
 
 MAX_JOB_COUNT = 1024
 
diff --git a/eventmq/scheduler.py b/eventmq/scheduler.py
index 2ff726a..e7c13ec 100644
--- a/eventmq/scheduler.py
+++ b/eventmq/scheduler.py
@@ -18,14 +18,15 @@
 Handles cron and other scheduled tasks
 """
 from hashlib import sha1 as emq_hash
+import importlib
 import json
 from json import dumps as serialize
 from json import loads as deserialize
 import logging
+import sys
 
 from croniter import croniter
-import redis
-from six import next
+from six import iteritems, next
 
 from eventmq.log import setup_logger
 
@@ -56,7 +57,6 @@ class Scheduler(HeartbeatMixin, EMQPService):
 
         logger.info('EventMQ Version {}'.format(__version__))
         logger.info('Initializing Scheduler...')
-        import_settings()
         super(Scheduler, self).__init__(*args, **kwargs)
         self.outgoing = Sender()
         self._redis_server = None
@@ -85,30 +85,28 @@ class Scheduler(HeartbeatMixin, EMQPService):
 
         self.poller = Poller()
 
-        self.load_jobs()
-
         self._setup()
 
     def load_jobs(self):
         """
-        Loads the jobs that need to be scheduled
+        Loads the jobs from redis that need to be scheduled
         """
-        try:
-            interval_job_list = self.redis_server.lrange(
-                'interval_jobs', 0, -1)
-            if interval_job_list is not None:
-                for i in interval_job_list:
-                    logger.debug('Restoring job with hash %s' % i)
-                    if (self.redis_server.get(i)):
-                        self.load_job_from_redis(
-                            message=deserialize(self.redis_server.get(i)))
-                    else:
-                        logger.warning('Expected scheduled job in redis,' +
-                                       'but none was found with hash %s' % i)
-        except redis.ConnectionError:
-            logger.warning('Could not contact redis server')
-        except Exception as e:
-            logger.warning(str(e))
+        if self.redis_server:
+            try:
+                interval_job_list = self.redis_server.lrange(
+                    'interval_jobs', 0, -1)
+                if interval_job_list is not None:
+                    for i in interval_job_list:
+                        logger.debug('Restoring job with hash %s' % i)
+                        if self.redis_server.get(i):
+                            self.load_job_from_redis(
+                                message=deserialize(self.redis_server.get(i)))
+                        else:
+                            logger.warning(
+                                'Expected scheduled job in redis, but none '
+                                'was found with hash {}'.format(i))
+            except Exception as e:
+                logger.warning(str(e), exc_info=True)
 
     def _start_event_loop(self):
         """
@@ -126,7 +124,6 @@ class Scheduler(HeartbeatMixin, EMQPService):
                 msg = self.outgoing.recv_multipart()
                 self.process_message(msg)
 
-            # TODO: distribute me!
             for hash_, cron in self.cron_jobs.items():
                 # If the time is now, or passed
                 if cron[0] <= ts_now:
@@ -145,10 +142,9 @@ class Scheduler(HeartbeatMixin, EMQPService):
                                  seconds_until(cron[0]))
 
             cancel_jobs = []
-            for k, v in self.interval_jobs.iteritems():
-                # TODO: Refactor this entire loop to be readable by humankind
-                # The schedule time has elapsed
+            for k, v in iteritems(self.interval_jobs):
                 if v[0] <= m_now:
+                    # The schedule time has elapsed
                     msg = v[1]
                     queue = v[3]
 
@@ -168,23 +164,25 @@ class Scheduler(HeartbeatMixin, EMQPService):
                                 v[0] = next(v[2])
                                 # Persist the change to redis if there are run
                                 # counts still left
-                                try:
-                                    message = deserialize(
-                                        self.redis_server.get(k))
-                                    new_headers = []
-                                    for header in message[1].split(','):
-                                        if 'run_count:' in header:
-                                            new_headers.append(
-                                                'run_count:{}'.format(v[4]))
-                                        else:
-                                            new_headers.append(header)
-                                    message[1] = ",".join(new_headers)
-                                    self.redis_server.set(
-                                        k, serialize(message))
-                                except Exception as e:
-                                    logger.warning(
-                                        'Unable to update key in redis '
-                                        'server: {}'.format(e))
+                                if self.redis_server:
+                                    try:
+                                        message = deserialize(
+                                            self.redis_server.get(k))
+                                        new_headers = []
+                                        for header in message[1].split(','):
+                                            if 'run_count:' in header:
+                                                new_headers.append(
+                                                    'run_count:{}'.format(
+                                                        v[4]))
+                                            else:
+                                                new_headers.append(header)
+                                        message[1] = ",".join(new_headers)
+                                        self.redis_server.set(
+                                            k, serialize(message))
+                                    except Exception as e:
+                                        logger.warning(
+                                            'Unable to update key in redis '
+                                            'server: {}'.format(e))
                             else:
                                 cancel_jobs.append(k)
 
@@ -204,21 +202,45 @@ class Scheduler(HeartbeatMixin, EMQPService):
 
     @property
     def redis_server(self):
-        # Open connection to redis server for persistance
         if self._redis_server is None:
+            # Open connection to redis server for persistence
+            cls_split = conf.REDIS_CLIENT_CLASS.split('.')
+            cls_path, cls_name = '.'.join(cls_split[:-1]), cls_split[-1]
             try:
-                self._redis_server = \
-                    redis.StrictRedis(host=conf.RQ_HOST,
-                                      port=conf.RQ_PORT,
-                                      db=conf.RQ_DB,
-                                      password=conf.RQ_PASSWORD)
-                return self._redis_server
+                mod = importlib.import_module(cls_path)
+                redis_cls = getattr(mod, cls_name)
+            except (ImportError, AttributeError) as e:
+                errmsg = 'Unable to import redis_client_class {} ({})'.format(
+                    conf.REDIS_CLIENT_CLASS, e)
+                logger.warning(errmsg)
+
+                if conf.REDIS_STARTUP_ERROR_HARD_KILL:
+                    sys.exit(1)
+                return None
+
+            url = 'redis://{}:{}/{}'.format(
+                conf.RQ_HOST, conf.RQ_PORT, conf.RQ_DB)
 
+            logger.info('Connecting to redis: {}{}'.format(
+                url, ' with password' if conf.RQ_PASSWORD else ''))
+
+            if conf.RQ_PASSWORD:
+                url = 'redis://:{}@{}:{}/{}'.format(
+                    conf.RQ_PASSWORD, conf.RQ_HOST, conf.RQ_PORT,
+                    conf.RQ_DB)
+
+            try:
+                self._redis_server = \
+                    redis_cls.from_url(url, **conf.REDIS_CLIENT_CLASS_KWARGS)
             except Exception as e:
                 logger.warning('Unable to connect to redis server: {}'.format(
                     e))
-        else:
-            return self._redis_server
+                if conf.REDIS_STARTUP_ERROR_HARD_KILL:
+                    sys.exit(1)
+
+            return None
+
+        return self._redis_server
 
     def send_request(self, jobmsg, queue=None):
         """
@@ -238,21 +260,28 @@ class Scheduler(HeartbeatMixin, EMQPService):
         return msgid
 
     def on_disconnect(self, msgid, message):
+        """Process request to shut down."""
         logger.info("Received DISCONNECT request: {}".format(message))
-        self._redis_server.connection_pool.disconnect()
         sendmsg(self.outgoing, KBYE)
         self.outgoing.unbind(conf.SCHEDULER_ADDR)
+
+        if self._redis_server:
+            # Check the internal var. No need to connect if we haven't already
+            # connected by this point
+            try:
+                self._redis_server.connection_pool.disconnect()
+            except Exception:
+                pass
         super(Scheduler, self).on_disconnect(msgid, message)
 
     def on_kbye(self, msgid, msg):
+        """Process router going offline"""
         if not self.is_heartbeat_enabled:
             self.reset()
 
     def on_unschedule(self, msgid, message):
-        """
-           Unschedule an existing schedule job, if it exists
-        """
-        logger.info("Received new UNSCHEDULE request: {}".format(message))
+        """Unschedule an existing schedule job, if it exists."""
+        logger.debug("Received new UNSCHEDULE request: {}".format(message))
 
         schedule_hash = self.schedule_hash(message)
         # TODO: Notify router whether or not this succeeds
@@ -263,11 +292,10 @@ class Scheduler(HeartbeatMixin, EMQPService):
         Cancels a job if it exists
 
         Args:
-            schedule_hash (str): The schedule's unique hash. See
-                :meth:`Scheduler.schedule_hash`
+            schedule_hash (str): The schedule's unique hash.
+                See :meth:`Scheduler.schedule_hash`
         """
         if schedule_hash in self.interval_jobs:
-
             # If the hash wasn't found in either `cron_jobs` or `interval_jobs`
             # then it's safe to assume it's already deleted.
             try:
@@ -282,18 +310,18 @@ class Scheduler(HeartbeatMixin, EMQPService):
         # Double check the redis server even if we didn't find the hash
         # in memory
         try:
-            if (self.redis_server.get(schedule_hash)):
-                self.redis_server.delete(schedule_hash)
+            if self.redis_server:
                 self.redis_server.lrem('interval_jobs', 0, schedule_hash)
-                self.redis_server.save()
-        except redis.ConnectionError as e:
-            logger.warning('Could not contact redis server: {}'.format(e))
+        except Exception as e:
+            logger.warning(str(e), exc_info=True)
+        try:
+            if self.redis_server and self.redis_server.get(schedule_hash):
+                self.redis_server.delete(schedule_hash)
         except Exception as e:
             logger.warning(str(e), exc_info=True)
 
     def load_job_from_redis(self, message):
-        """
-        """
+        """Parses and loads a message from redis as a scheduler job."""
         from .utils.timeutils import IntervalIter
 
         queue = message[0].encode('utf-8')
@@ -329,8 +357,7 @@ class Scheduler(HeartbeatMixin, EMQPService):
             self.cron_jobs[schedule_hash] = [c_next, message[3], c, queue]
 
     def on_schedule(self, msgid, message):
-        """
-        """
+        """Create a new scheduled job."""
         logger.info("Received new SCHEDULE request: {}".format(message))
 
         queue = message[0]
@@ -380,19 +407,16 @@ class Scheduler(HeartbeatMixin, EMQPService):
                 self.interval_jobs.pop(schedule_hash)
 
         # Persist the scheduled job
-        try:
-            if schedule_hash not in self.redis_server.lrange(
-                    'interval_jobs', 0, -1):
-                self.redis_server.lpush('interval_jobs', schedule_hash)
-            self.redis_server.set(schedule_hash, serialize(message))
-            self.redis_server.save()
-            logger.debug('Saved job {} with hash {} to redis'.format(
-                message, schedule_hash))
-        except redis.ConnectionError:
-            logger.warning('Could not contact redis server. Unable to '
-                           'guarantee persistence.')
-        except Exception as e:
-            logger.warning(str(e))
+        if self.redis_server:
+            try:
+                if schedule_hash not in self.redis_server.lrange(
+                        'interval_jobs', 0, -1):
+                    self.redis_server.lpush('interval_jobs', schedule_hash)
+                    self.redis_server.set(schedule_hash, serialize(message))
+                logger.debug('Saved job {} with hash {} to redis'.format(
+                    message, schedule_hash))
+            except Exception as e:
+                logger.warning(str(e))
 
         # Send a request in haste mode, decrement run_count if needed
         if 'nohaste' not in headers:
@@ -410,21 +434,17 @@ class Scheduler(HeartbeatMixin, EMQPService):
         return run_count
 
     def on_heartbeat(self, msgid, message):
-        """
-        Noop command. The logic for heartbeating is in the event loop.
-        """
+        """Noop command. The logic for heart beating is in the event loop."""
 
     @classmethod
     def schedule_hash(cls, message):
-        """
-        Create a unique identifier for this message for storing
-        and referencing later
+        """Create a unique identifier to store and reference a message later.
 
         Args:
             message (str): The serialized message passed to the scheduler
 
         Returns:
-            int: unique hash for the job
+            str: unique hash for the job
         """
 
         # Get the job portion of the message
@@ -452,18 +472,17 @@ class Scheduler(HeartbeatMixin, EMQPService):
         """
         setup_logger("eventmq")
         import_settings()
-        self.__init__()
-        self.start(addr=conf.SCHEDULER_ADDR)
-
+        import_settings('scheduler')
 
-# Entry point for pip console scripts
-def scheduler_main():
-    s = Scheduler()
-    s.scheduler_main()
+        self.load_jobs()
+        self.start(addr=conf.SCHEDULER_ADDR)
 
 
 def test_job(*args, **kwargs):
     """
     Simple test job for use with the scheduler
     """
+    from pprint import pprint
     print("hello!")  # noqa
+    pprint(args)  # noqa
+    pprint(kwargs)  # noqa
diff --git a/setup.py b/setup.py
index 1427d67..dceac7f 100644
--- a/setup.py
+++ b/setup.py
@@ -22,7 +22,6 @@ setup(
                       'six==1.10.0',
                       'monotonic==0.4',
                       'croniter==0.3.10',
-                      'redis==2.10.3',
                       'future==0.15.2',
                       'psutil==5.0.0',
                       'python-dateutil>=2.1,<3.0.0'],