code: 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))