Redis EventSourcing sub.

    This sub should be used together with sqlalchemy_es_pub, it will
    use RedisEventStore as events storage layer and use the prepare-commit
    pattern in :func:`sqlalchemy_es_pub` to ensure 100% security on
    events recording.

    :param session: the sqlalchemy to bind the signal
    :param tables: tables to be event sourced.
    :param redis_dsn: the redis server to store event sourcing events.
    :param strict: arg to be passed to RedisPrepareCommit. If set to True,
     the exception will not be silent and may cause the failure of sqlalchemy
     transaction, user should handle the exception in the app side in this
     case.
    :param namespace: namespace string or func. If func passed, it should
     accept timestamp as arg and return a string namespace.
    :param ttl: expiration time for events stored, default to 3 days.
    :param socket_timeout: redis socket timeout.

def redis_es_sub(session, tables, redis_dsn, strict=False,
                 namespace=None, ttl=3600*24*3, socket_timeout=1):
    """Redis EventSourcing sub.

    This sub should be used together with sqlalchemy_es_pub, it will
    use RedisEventStore as events storage layer and use the prepare-commit
    pattern in :func:`sqlalchemy_es_pub` to ensure 100% security on
    events recording.

    :param session: the sqlalchemy to bind the signal
    :param tables: tables to be event sourced.
    :param redis_dsn: the redis server to store event sourcing events.
    :param strict: arg to be passed to RedisPrepareCommit. If set to True,
     the exception will not be silent and may cause the failure of sqlalchemy
     transaction, user should handle the exception in the app side in this
     case.
    :param namespace: namespace string or func. If func passed, it should
     accept timestamp as arg and return a string namespace.
    :param ttl: expiration time for events stored, default to 3 days.
    :param socket_timeout: redis socket timeout.
    """
    logger = logging.getLogger("meepo.sub.redis_es_sub")

    if not isinstance(tables, (list, set)):
        raise ValueError("tables should be list or set")

    # install event store hook for tables
    event_store = RedisEventStore(
        redis_dsn, namespace=namespace, ttl=ttl, socket_timeout=socket_timeout)

    def _es_event_sub(pk, event):
        if event_store.add(event, str(pk)):
            logger.info("%s: %s -> %s" % (
                event, pk, datetime.datetime.now()))
        else:
            logger.error("event sourcing failed: %s" % pk)

    events = ("%s_%s" % (tb, action) for tb, action in
              itertools.product(*[tables, ["write", "update",  "delete"]]))
    for event in events:
        sub_func = functools.partial(_es_event_sub, event=event)
        signal(event).connect(sub_func, weak=False)

    # install prepare-commit hook
    prepare_commit = RedisPrepareCommit(
        redis_dsn, strict=strict, namespace=namespace,
        socket_timeout=socket_timeout)

    signal("session_prepare").connect(
        prepare_commit.prepare, sender=session, weak=False)
    signal("session_commit").connect(
        prepare_commit.commit, sender=session, weak=False)
    signal("session_rollback").connect(
        prepare_commit.rollback, sender=session, weak=False)

    return event_store, prepare_commit