def mysql_pub(mysql_dsn, tables=None, blocking=False, **kwargs):
    parsed = urlparse(mysql_dsn)
    mysql_settings = {
        "host": parsed.hostname,
        "port": parsed.port or 3306,
        "user": parsed.username,
        "passwd": parsed.password
    }
    stream = pymysqlreplication.BinLogStreamReader(
        mysql_settings,
        server_id=random.randint(1000000000, 4294967295),
        blocking=blocking,
        only_events=[DeleteRowsEvent, UpdateRowsEvent, WriteRowsEvent],
        **kwargs
    )
    def _pk(values):
        if isinstance(event.primary_key, str):
            return values[event.primary_key]
        return tuple(values[k] for k in event.primary_key)
    for event in stream:
        if not event.primary_key:
            continue
        if tables and event.table not in tables:
            continue
        try:
            rows = event.rows
        except (UnicodeDecodeError, ValueError) as e:
            logger.exception(e)
            continue
        timestamp = datetime.datetime.fromtimestamp(event.timestamp)
        if isinstance(event, WriteRowsEvent):
            sg_name = "%s_write" % event.table
            sg = signal(sg_name)
            sg_raw = signal("%s_raw" % sg_name)
            for row in rows:
                pk = _pk(row["values"])
                sg.send(pk)
                sg_raw.send(row)
                logger.debug("%s -> %s, %s" % (sg_name, pk, timestamp))
        elif isinstance(event, UpdateRowsEvent):
            sg_name = "%s_update" % event.table
            sg = signal(sg_name)
            sg_raw = signal("%s_raw" % sg_name)
            for row in rows:
                pk = _pk(row["after_values"])
                sg.send(pk)
                sg_raw.send(row)
                logger.debug("%s -> %s, %s" % (sg_name, pk, timestamp))
        elif isinstance(event, DeleteRowsEvent):
            sg_name = "%s_delete" % event.table
            sg = signal(sg_name)
            sg_raw = signal("%s_raw" % sg_name)
            for row in rows:
                pk = _pk(row["values"])
                sg.send(pk)
                sg_raw.send(row)
                logger.debug("%s -> %s, %s" % (sg_name, pk, timestamp))
        signal("mysql_binlog_pos").send(
            "%s:%s" % (stream.log_file, stream.log_pos))