Merge pull request #41 from synic/debounce0.2.5

Add debounce function

9 files changed, 626 insertions, 128 deletions

diff --git a/circle.yml b/circle.yml
index dfdc164..bbbdb47 100644
--- a/circle.yml
+++ b/circle.yml
@@ -1,3 +1,6 @@
+machine:
+  services:
+    - redis
 test:
   override:
     - pip install --upgrade -r requirements_.txt
diff --git a/eventmq/client/debounce.py b/eventmq/client/debounce.py
new file mode 100644
index 0000000..e6f462e
--- /dev/null
+++ b/eventmq/client/debounce.py
@@ -0,0 +1,295 @@
+# This file is part of eventmq.
+#
+# eventmq is free software: you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation, either version 2.1 of the License, or (at your option)
+# any later version.
+#
+# eventmq is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with eventmq.  If not, see <http://www.gnu.org/licenses/>.
+import logging
+import copy
+
+from .. import conf
+
+logger = logging.getLogger(__name__)
+
+# attempt to import redis
+try:
+    import redis  # noqa
+except ImportError:
+    redis = None
+    logger.warning('Redis not installed, debounce support not available.')
+
+
+def _debounce_run_deferred_job(context):
+    """Called by the debounce system to actually run the job.
+
+    This is called by the debounce system at the end of the process, when the
+    job should actually run.  It deserializes the callable, the arguments, and
+    tries to execute the callable with the arguments.
+
+    context keys:
+        path (str): The path of the callable function (the module information)
+        callable_name (str): The name of the callable
+        args (list): The arguments to pass when invoking the callable
+        kwargs (dict): The arguments to pass when invoking the callable
+        class_args (list): Arguments to use when initializing the callable
+            class (if it is a class)
+        class_kwargs (dict): The arguments to use when initializing the
+            callable class (if it is a class)
+        cache_key1 (str): The cache key for step 1
+        cache_key2 (str): The cache key for step 2
+        queue (str): The queue name for this job
+        job_id (str): The unique job identifier
+        scheduled (bool): Whether or not this was deferred immediately, or
+            scheduled for the future.
+
+    Args:
+       context (dict): See above for "context keys"
+    """
+    from ..sender import Sender
+    from ..utils.functions import run_function
+    from .messages import schedule
+
+    context = copy.deepcopy(context)  # don't modify caller's context
+    queue = context.get('queue')
+    job_id = context.get('job_id')
+    path = context.get('path')
+    callable_name = context.get('callable_name')
+    scheduled = context.get('scheduled')
+    cache_key1 = context.get('cache_key1')
+    cache_key2 = context.get('cache_key2')
+    args = context.get('args')
+    kwargs = context.get('kwargs')
+    class_args = context.get('class_args')
+    class_kwargs = context.get('class_kwargs')
+
+    if queue:
+        queue = str(queue)
+
+    if job_id:
+        job_id = str(job_id)
+
+    logger.debug('DEBOUNCE: {}.{} - running {} job'.format(
+        path, callable_name, 'scheduled' if job_id else 'deferred'))
+
+    if scheduled and job_id:
+        # TODO: here we create a new socket just so we can unschedule the
+        # debunce scheduled job.  This can be removed once the `run_x` feature
+        # is implemented that lets you specify how many times a scheduled job
+        # should run before stopping.
+        socket = Sender()
+        socket.connect(addr=conf.FRONTEND_ADDR)
+        schedule(socket, _debounce_run_deferred_job,
+                 queue=queue, unschedule=True, class_args=(job_id, ))
+        socket.zsocket.close()
+
+        del socket
+
+    try:
+        redis_connection = redis.StrictRedis(
+            host=conf.RQ_HOST,
+            port=conf.RQ_PORT,
+            db=conf.RQ_DB,
+            password=conf.RQ_PASSWORD)
+    except Exception as e:
+        logger.error('DEBOUNCE: {}.{} - requested, but could not connect '
+                     'to redis server'.format(path, callable_name, e))
+        redis_connection.delete(cache_key1)
+        if scheduled:
+            redis_connection.delete(cache_key2)
+        return
+
+    # if the timer countdown caused this function to run, and `cache_key1` is
+    # set, bail out.  Otherwise, continue, because the cache is set prior to
+    # deferring the function.
+    if redis_connection.get(cache_key1) and scheduled:
+        logger.debug('DEBOUNCE: {}.{} - cache_key1 was set, not '
+                     'running job.'.format(path, callable_name))
+        return
+
+    redis_connection.set(cache_key1, 1)
+    redis_connection.expire(
+        cache_key1, getattr(conf, 'DEBOUNCE_CACHE_KEY1_TIMEOUT', 60))
+
+    try:
+        run_function(
+            path, callable_name, class_args, class_kwargs, *args, **kwargs)
+    except Exception as e:
+        logger.error('DEBOUNCE: {}.{} - exception {}'.format(
+            path, callable_name, e))
+
+    redis_connection.delete(cache_key1)
+    if scheduled:
+        redis_connection.delete(cache_key2)
+
+
+def _debounce_deferred_job(context):
+    """Debounce the job.
+
+    context keys:
+        path (str): The path of the callable function (the module information)
+        callable_name (str): The name of the callable
+        args (list): The arguments to pass when invoking the callable
+        kwargs (dict): The arguments to pass when invoking the callable
+        class_args (list): Arguments to use when initializing the callable
+            class (if it is a class)
+        class_kwargs (dict): The arguments to use when initializing the
+            callable class (if it is a class)
+        queue (str): The queue name for this job
+        debounce_secs (int): Debounce schedule interval, in seconds
+
+    Args:
+        context (dict): See "context keys" above.
+    """
+    from ..sender import Sender
+    from ..utils.functions import arguments_hash
+    from .messages import schedule
+
+    job_id = arguments_hash(locals())
+    context = copy.deepcopy(context)  # don't modify caller's context
+    path = context.get('path')
+    callable_name = context.get('callable_name')
+    debounce_secs = context.pop('debounce_secs')
+
+    if not redis:
+        logger.error('DEBOUNCE requested, but redis is not available.')
+        return
+
+    try:
+        redis_connection = redis.StrictRedis(
+            host=conf.RQ_HOST,
+            port=conf.RQ_PORT,
+            db=conf.RQ_DB,
+            password=conf.RQ_PASSWORD)
+    except Exception as e:
+        logger.error('Debounce requested, but could not connect to '
+                     'redis server: {}.'.format(e))
+        return
+
+    cache_key1 = '{}_key1'.format(job_id)
+    cache_key2 = '{}_key2'.format(job_id)
+
+    logger.debug('DEBOUNCE: {}.{} - job id: {}'.format(
+        path, callable_name, job_id))
+
+    context.update({
+        'cache_key1': cache_key1,
+        'cache_key2': cache_key2,
+        'scheduled': False,
+        })
+
+    if not redis_connection.get(cache_key1):
+        logger.debug('DEBOUNCE: {}.{} - cache_key1 was not set, '
+                     'deferring immediately.'.format(path, callable_name))
+        redis_connection.set(cache_key1, 1)
+        redis_connection.expire(
+            cache_key1, getattr(conf, 'DEBOUNCE_CACHE_KEY1_TIMEOUT', 60))
+
+        _debounce_run_deferred_job(context)
+        return
+    else:
+        logger.debug(
+            'DEBOUNCE: {}.{} - cache_key1 was set, '
+            'trying cache_key2...'.format(path, callable_name))
+
+    if not redis_connection.get(cache_key2):
+        logger.debug('DEBOUNCE: {}.{} - cache_key2 was not set, scheduling '
+                     'for {} seconds'.format(path, callable_name,
+                                             debounce_secs))
+        redis_connection.set(cache_key2, 1)
+        redis_connection.expire(cache_key2, debounce_secs)
+
+        context['job_id'] = job_id
+        context['scheduled'] = True
+
+        # TODO: here we create a new socket just so we can schedule the
+        # debunce scheduled job.  This should probably use the socket that was
+        # passed to the original `defer_job` call, however, this is not
+        # currently possible in eventmq.
+        socket = Sender()
+        socket.connect(addr=conf.FRONTEND_ADDR)
+
+        schedule(
+            socket=socket,
+            func=_debounce_run_deferred_job,
+            queue=context.get('queue'),
+            interval_secs=debounce_secs,
+            kwargs={'context': context},
+            headers=('nohaste', 'guarantee'),
+            class_args=(job_id, ),
+        )
+        socket.zsocket.close()
+
+        del socket
+        return
+    else:
+        logger.debug(
+            'DEBOUNCE: {}.{} - cache_key2 was set, dropping message.  '.format(
+                path, callable_name))
+
+
+def _debounce_schedule(
+        socket, path, callable_name, debounce_secs, args=(), kwargs=None,
+        class_args=(), class_kwargs=None, reply_requested=False,
+        guarantee=False, retry_count=0, queue=conf.DEFAULT_QUEUE_NAME):
+    """Schedule the initial debounce.
+
+    Debounce works like this:
+
+    1.  When the first job comes in, `cache_key1` will not be set, so it
+    will defer the job to run immediately.  At the beginning of that job,
+    it will set `cache_key1`, and at the end, it will unset `cache_key1`.
+
+    2.  If `cache_key1` is set and a new job comes in, if `cache_key2` is
+    not set, it will set `cache_key2` and schedule the job to run
+    `debounce_secs` in the future.
+
+    3.  If a third job comes in while `cache_key1` and `cache_key2` are
+    set, it will be ignored.
+
+    Args:
+        path (str): The path of the callable function (the module information)
+        callable_name (str): The name of the callable
+        debounce_secs (int): Debounce schedule interval, in seconds
+        reply_requested (bool): request the return value of func as a reply
+        retry_count (int): How many times should be retried when encountering
+            an Exception or some other failure before giving up. (default: 0
+            or immediately fail)
+        queue (str): Name of queue to use when executing the job. If this value
+            evaluates to False, the default is used. Default: is configured
+            default queue name
+    """
+
+    from .messages import defer_job
+
+    logger.debug('DEBOUNCE: {}.{} - initial schedule'.format(
+        path, callable_name))
+
+    context = {
+        'path': path,
+        'callable_name': callable_name,
+        'args': args,
+        'kwargs': kwargs,
+        'class_args': class_args,
+        'class_kwargs': class_kwargs,
+        'queue': queue,
+        'scheduled': False,
+        'debounce_secs': debounce_secs,
+    }
+
+    defer_job(
+        socket,
+        _debounce_deferred_job,
+        reply_requested=reply_requested,
+        retry_count=retry_count,
+        queue=queue,
+        guarantee=guarantee,
+        kwargs={'context': context},
+        )
diff --git a/eventmq/client/messages.py b/eventmq/client/messages.py
index cd26b3f..e589d4d 100644
--- a/eventmq/client/messages.py
+++ b/eventmq/client/messages.py
@@ -16,21 +16,16 @@
 :mod:`messages` -- Client Messaging
 ===================================
 """
-import inspect
 import logging
-import importlib
 from json import dumps as serialize
 
 from .. import conf
 from ..utils.messages import send_emqp_message
+from ..utils.functions import path_from_callable
 
 logger = logging.getLogger(__name__)
 
 
-class CallableFromPathError(Exception):
-    pass
-
-
 def schedule(socket, func, interval_secs=None, args=(), kwargs=None,
              class_args=(), class_kwargs=None, headers=('guarantee',),
              queue=conf.DEFAULT_QUEUE_NAME, unschedule=False, cron=None):
@@ -111,9 +106,10 @@ def schedule(socket, func, interval_secs=None, args=(), kwargs=None,
     return msgid
 
 
-def defer_job(socket, func, args=(), kwargs=None, class_args=(),
-              class_kwargs=None, reply_requested=False, guarantee=False,
-              retry_count=0, queue=conf.DEFAULT_QUEUE_NAME):
+def defer_job(
+        socket, func, args=(), kwargs=None, class_args=(), class_kwargs=None,
+        reply_requested=False, guarantee=False, retry_count=0,
+        debounce_secs=False, queue=conf.DEFAULT_QUEUE_NAME):
     """
     Used to send a job to a worker to execute via `socket`.
 
@@ -130,11 +126,11 @@ def defer_job(socket, func, args=(), kwargs=None, class_args=(),
         class_kwargs (dict): dict of **kwargs to pass to the class when
             initializing (if applicable).
         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 immediately fail)
+        debounce_secs (secs): Number of seconds to debounce the job.   See
+            `debounce_deferred_job` for more information.
         queue (str): Name of queue to use when executing the job. If this value
             evaluates to False, the default is used. Default: is configured
             default queue name
@@ -142,6 +138,8 @@ def defer_job(socket, func, args=(), kwargs=None, class_args=(),
         str: ID for the message/deferred job. This value will be None if there
             was an error.
     """
+    from eventmq.client import debounce
+
     callable_name = None
     path = None
 
@@ -172,6 +170,25 @@ def defer_job(socket, func, args=(), kwargs=None, class_args=(),
                      format(func.__name__))
         return
 
+    if debounce_secs:
+        logger.debug('DEBOUNCE: {} - called, debounce_secs: {}...'.format(
+            func.__name__,
+            debounce_secs))
+        debounce._debounce_schedule(
+            socket=socket,
+            path=path,
+            callable_name=callable_name,
+            debounce_secs=debounce_secs,
+            args=args,
+            kwargs=kwargs,
+            class_args=class_args,
+            class_kwargs=class_kwargs,
+            reply_requested=reply_requested,
+            guarantee=guarantee,
+            retry_count=retry_count,
+            queue=queue)
+        return
+
     msg = ['run', {
         'callable': callable_name,
         'path': path,
@@ -190,100 +207,6 @@ def defer_job(socket, func, args=(), kwargs=None, class_args=(),
     return msgid
 
 
-def path_from_callable(func):
-    """
-    Builds the module path in string format for a callable.
-
-    .. note:
-       To use a callable Object, pass Class.__call__ as func and provide any
-       class_args/class_kwargs. This is so side effects from pickling won't
-       occur.
-
-    Args:
-        func (callable): The function or method to build the path for
-
-    Returns:
-        list: (import path (w/ class seperated by a ':'), callable name) or
-            (None, None) on error.
-    """
-    callable_name = None
-
-    path = None
-    # Methods also have the func_name property
-    if inspect.ismethod(func):
-        path = ("{}:{}".format(func.__module__, func.im_class.__name__))
-        callable_name = func.func_name
-    elif inspect.isfunction(func):
-        path = func.__module__
-        callable_name = func.func_name
-    else:
-        # We should account for another callable type so log information
-        # about it
-        if hasattr(func, '__class__') and isinstance(func, func.__class__):
-            func_type = 'instanceobject'
-        else:
-            func_type = type(func)
-
-        logger.error('Encountered unknown callable ({}) type {}'.format(
-            func,
-            func_type
-        ))
-        return None, None
-
-    return path, callable_name
-
-
-def callable_from_path(path, callable_name, *args, **kwargs):
-    """Build a callable from a path and callable_name.
-
-    This function is the opposite of `path_from_callable`.  It takes what is
-    returned from `build_module_name` and converts it back to the original
-    caller.
-
-    Args:
-        path (str): The module path of the callable.  This is the first
-            position in the tuple returned from `path_from_callable`.
-        callable_name (str): The name of the function.  This is the second
-            position of the tuple returned from `path_from_callable`.
-        *args (list): if `callable_name` is a method on a class, these
-            arguments will be passed to the constructor when instantiating the
-            class.
-        *kwargs (dict): if `callable_name` is a method on a class, these
-            arguments will be passed to the constructor when instantiating the
-            class.
-
-    Returns:
-        function: The callable
-    """
-    if ':' in path:
-        _pksplit = path.split(':')
-        s_package = _pksplit[0]
-        s_cls = _pksplit[1]
-    else:
-        s_package = path
-        s_cls = None
-
-    try:
-        package = importlib.import_module(s_package)
-        reload(package)
-    except Exception as e:
-        raise CallableFromPathError(str(e))
-
-    if s_cls:
-        cls = getattr(package, s_cls)
-
-        obj = cls(*args, **kwargs)
-    else:
-        obj = package
-
-    try:
-        callable_ = getattr(obj, callable_name)
-    except AttributeError as e:
-        raise CallableFromPathError(str(e))
-
-    return callable_
-
-
 def send_request(socket, message, reply_requested=False, guarantee=False,
                  retry_count=0, queue=None):
     """
diff --git a/eventmq/conf.py b/eventmq/conf.py
index 0c4b7cc..59c8f90 100644
--- a/eventmq/conf.py
+++ b/eventmq/conf.py
@@ -20,7 +20,7 @@
 #: SUPER_DEBUG basically enables more debugging logs. Specifically the messages
 #: at different levels in the application.
 #: Default: False
-SUPER_DEBUG = False
+SUPER_DEBUG = True
 
 #: Don't show HEARTBEAT message when debug logging is enabled
 #: Default: True
diff --git a/eventmq/exceptions.py b/eventmq/exceptions.py
index 955709f..1277cd2 100644
--- a/eventmq/exceptions.py
+++ b/eventmq/exceptions.py
@@ -55,3 +55,10 @@ class UnknownQueueError(EventMQError):
     """
     Raised when a queue is not found in the internal list of queues.
     """
+
+
+class CallableFromPathError(EventMQError):
+    """
+    Raised when construction of a callable from a path and callable_name fails.
+    """
+    pass
diff --git a/eventmq/tests/test_client_messages.py b/eventmq/tests/test_client_messages.py
index 49d7ca2..3337987 100644
--- a/eventmq/tests/test_client_messages.py
+++ b/eventmq/tests/test_client_messages.py
@@ -105,7 +105,7 @@ class TestCase(unittest.TestCase):
                 ('eventmq.client.messages',
                  'ERROR',
                  'Encountered callable with no __module__ path nameless_func'),
-                ('eventmq.client.messages',
+                ('eventmq.utils.functions',
                  'ERROR',
                  'Encountered unknown callable ({}) type instanceobject'.
                  format(callable_obj)),
@@ -210,7 +210,7 @@ class TestCase(unittest.TestCase):
                 ('eventmq.client.messages',
                  'ERROR',
                  'Encountered callable with no __module__ path nameless_func'),
-                ('eventmq.client.messages',
+                ('eventmq.utils.functions',
                  'ERROR',
                  'Encountered unknown callable ({}) type instanceobject'.
                  format(callable_obj)),
diff --git a/eventmq/tests/test_debounce.py b/eventmq/tests/test_debounce.py
new file mode 100644
index 0000000..15008ab
--- /dev/null
+++ b/eventmq/tests/test_debounce.py
@@ -0,0 +1,134 @@
+import uuid
+import unittest
+import mock
+import redis
+
+from .. import conf
+
+from eventmq.client import messages
+
+
+def test_job_function():
+    return 'test_job_function'
+
+
+class TestCase(unittest.TestCase):
+    def setUp(self):
+        super(TestCase, self).setUp()
+        self.redis = redis.StrictRedis(
+            host=conf.RQ_HOST,
+            port=conf.RQ_PORT,
+            db=conf.RQ_DB,
+            password=conf.RQ_PASSWORD)
+        self.socket = mock.Mock()
+
+        self.job_id = str(uuid.uuid4())
+        self.arguments_hash_mock = mock.patch(
+            'eventmq.utils.functions.arguments_hash', lambda *x: self.job_id)
+        self.arguments_hash_mock.start()
+
+    def tearDown(self):
+        cache_key1 = '{}_key1'.format(self.job_id)
+        cache_key2 = '{}_key2'.format(self.job_id)
+        self.redis.delete(cache_key1)
+        self.redis.delete(cache_key2)
+
+        self.arguments_hash_mock.stop()
+
+        super(TestCase, self).tearDown()
+
+    @mock.patch('eventmq.client.debounce._debounce_schedule')
+    def test_debounce_initial_call(self, debounce_schedule_mock):
+        messages.defer_job(
+            self.socket,
+            test_job_function,
+            debounce_secs=1)
+
+        debounce_schedule_mock.assert_called_with(
+            path='eventmq.tests.test_debounce',
+            callable_name='test_job_function',
+            args=(),
+            kwargs={},
+            class_args=(),
+            class_kwargs={},
+            queue='default',
+            reply_requested=False,
+            retry_count=0,
+            guarantee=False,
+            debounce_secs=1,
+            socket=self.socket)
+
+    @mock.patch('eventmq.client.debounce._debounce_run_deferred_job')
+    def test_debounce_step1(self, run_deferred_job_mock):
+        """Should call through to _debounce_run_deferred_job."""
+        from eventmq.client import debounce
+        from eventmq.utils.functions import path_from_callable
+
+        path, callable_name = path_from_callable(test_job_function)
+
+        context = {
+            'job_id': self.job_id,
+            'cache_key1': '{}_key1'.format(self.job_id),
+            'cache_key2': '{}_key2'.format(self.job_id),
+            'path': path,
+            'callable_name': callable_name,
+            'queue': 'default',
+            'debounce_secs': 1,
+            'scheduled': False,
+        }
+
+        debounce._debounce_deferred_job(context)
+
+        context.pop('debounce_secs', None)
+
+        run_deferred_job_mock.assert_called_with(context)
+
+    @mock.patch('eventmq.sender.Sender')
+    @mock.patch('eventmq.client.debounce._debounce_run_deferred_job')
+    @mock.patch('eventmq.client.messages.schedule')
+    def test_debounce_step2(self, schedule_mock, run_deferred_job_mock,
+                            sender_mock):
+        """Should call through to _debounce_run_deferred_job."""
+        from eventmq.client import debounce
+        from eventmq.utils.functions import path_from_callable
+
+        sender_mock.return_value = self.socket
+
+        path, callable_name = path_from_callable(test_job_function)
+
+        context = {
+            'job_id': self.job_id,
+            'cache_key1': '{}_key1'.format(self.job_id),
+            'cache_key2': '{}_key2'.format(self.job_id),
+            'path': path,
+            'callable_name': callable_name,
+            'queue': 'default',
+            'debounce_secs': 1,
+            'scheduled': False,
+        }
+
+        debounce._debounce_deferred_job(context)
+
+        context.pop('debounce_secs', None)
+
+        run_deferred_job_mock.assert_called_with(context)
+        run_deferred_job_mock.reset_mock()
+
+        # this time, since we already went through step1, it should schedule a
+        # job for step2
+        context['debounce_secs'] = 1
+        debounce._debounce_deferred_job(context)
+
+        run_deferred_job_mock.assert_not_called()
+
+        context['scheduled'] = True
+        context.pop('debounce_secs', None)
+
+        schedule_mock.assert_called_with(
+            socket=self.socket,
+            func=run_deferred_job_mock,
+            headers=('nohaste', 'guarantee'),
+            class_args=(self.job_id,),
+            interval_secs=1,
+            kwargs={'context': context},
+            queue='default')
diff --git a/eventmq/utils/functions.py b/eventmq/utils/functions.py
new file mode 100644
index 0000000..0bd1ba9
--- /dev/null
+++ b/eventmq/utils/functions.py
@@ -0,0 +1,151 @@
+import json
+import hashlib
+import importlib
+import inspect
+
+from .. import log
+from ..exceptions import CallableFromPathError
+
+logger = log.setup_logger(__name__)
+
+
+class IgnoreJSONEncoder(json.JSONEncoder):
+    """JSON Encoder that ignores unknown keys."""
+    def default(self, obj):
+        try:
+            return super(IgnoreJSONEncoder, self).default(obj)
+        except TypeError:
+            return None
+
+
+def run_function(path, callable_name,
+                 class_args=(), class_kwargs=None, *args, **kwargs):
+    """Constructs a callable from `path` and `callable_name` and calls it.
+
+    Args:
+        class_args (list): If the callable is a class method, these args will
+            be used to initialize the class.
+        class_kwargs (dict): If the callable is a class method, these kwargs
+            will be used to initialize the class.
+        args (list): These arguments will be passed as parameters to the
+            callable
+        kwargs (dict): These keyword arguments will be passed as parameters to
+            the callable.
+
+    Return:
+        object: Whatever is returned by calling the callable will be returned
+            from this function.
+    """
+    try:
+        callable_ = callable_from_path(
+            path, callable_name, *class_args, **class_kwargs)
+    except CallableFromPathError as e:
+        logger.exception('Error importing callable {}.{}: {}'.format(
+            path, callable_name, str(e)))
+        return
+
+    return callable_(*args, **kwargs)
+
+
+def arguments_hash(*args, **kwargs):
+    """Takes `args` and `kwargs` and creates a unique identifier."""
+    args = {
+        'args': args,
+        'kwargs': kwargs,
+    }
+
+    data = json.dumps(args, cls=IgnoreJSONEncoder)
+    return hashlib.sha1(data).hexdigest()
+
+
+def path_from_callable(func):
+    """
+    Builds the module path in string format for a callable.
+
+    .. note:
+       To use a callable Object, pass Class.__call__ as func and provide any
+       class_args/class_kwargs. This is so side effects from pickling won't
+       occur.
+
+    Args:
+        func (callable): The function or method to build the path for
+
+    Returns:
+        list: (import path (w/ class seperated by a ':'), callable name) or
+            (None, None) on error.
+    """
+    callable_name = None
+
+    path = None
+    # Methods also have the func_name property
+    if inspect.ismethod(func):
+        path = ("{}:{}".format(func.__module__, func.im_class.__name__))
+        callable_name = func.func_name
+    elif inspect.isfunction(func):
+        path = func.__module__
+        callable_name = func.func_name
+    else:
+        # We should account for another callable type so log information
+        # about it
+        if hasattr(func, '__class__') and isinstance(func, func.__class__):
+            func_type = 'instanceobject'
+        else:
+            func_type = type(func)
+
+        logger.error('Encountered unknown callable ({}) type {}'.format(
+            func,
+            func_type
+        ))
+        return None, None
+
+    return path, callable_name
+
+
+def callable_from_path(path, callable_name, *args, **kwargs):
+    """Build a callable from a path and callable_name.
+
+    This function is the opposite of `path_from_callable`.  It takes what is
+    returned from `build_module_name` and converts it back to the original
+    caller.
+
+    Args:
+        path (str): The module path of the callable.  This is the first
+            position in the tuple returned from `path_from_callable`.
+        callable_name (str): The name of the function.  This is the second
+            position of the tuple returned from `path_from_callable`.
+        *args (list): if `callable_name` is a method on a class, these
+            arguments will be passed to the constructor when instantiating the
+            class.
+        *kwargs (dict): if `callable_name` is a method on a class, these
+            arguments will be passed to the constructor when instantiating the
+            class.
+
+    Returns:
+        function: The callable
+    """
+    if ':' in path:
+        _pksplit = path.split(':')
+        s_package = _pksplit[0]
+        s_cls = _pksplit[1]
+    else:
+        s_package = path
+        s_cls = None
+
+    try:
+        package = importlib.import_module(s_package)
+        reload(package)
+    except Exception as e:
+        raise CallableFromPathError(str(e))
+
+    if s_cls:
+        cls = getattr(package, s_cls)
+        obj = cls(*args, **kwargs)
+    else:
+        obj = package
+
+    try:
+        callable_ = getattr(obj, callable_name)
+    except AttributeError as e:
+        raise CallableFromPathError(str(e))
+
+    return callable_
diff --git a/eventmq/worker.py b/eventmq/worker.py
index 852c7da..6b1773b 100644
--- a/eventmq/worker.py
+++ b/eventmq/worker.py
@@ -17,7 +17,7 @@
 ===============================
 Defines different short-lived workers that execute jobs
 """
-from client.messages import callable_from_path, CallableFromPathError
+from .utils.functions import run_function
 
 # the run function is executed in a different process, so we need to set the
 # logger up.
@@ -37,27 +37,12 @@ def run(payload, msgid):
     callable_name = payload.get('callable')
     class_args = payload.get('class_args', tuple()) or tuple()
     class_kwargs = payload.get('class_kwargs', dict()) or dict()
+    args = payload.get('args', tuple()) or tuple()
+    kwargs = payload.get('kwargs', dict()) or dict()
 
     try:
-        callable_ = callable_from_path(
-            path, callable_name, *class_args, **class_kwargs)
-    except CallableFromPathError as e:
-        logger.exception('Error importing callable {}.{}: {}'.format(
-            path, callable_name, str(e)))
-        return (msgid, str(e))
-
-    if "args" in payload:
-        args = payload["args"]
-    else:
-        args = ()
-
-    if "kwargs" in payload:
-        kwargs = payload["kwargs"]
-    else:
-        kwargs = {}
-
-    try:
-        r = callable_(*args, **kwargs)
+        r = run_function(
+            path, callable_name, class_args, class_kwargs, *args, **kwargs)
         return (msgid, r)
     except Exception as e:
         logger.exception(e)