def mysql_stream(conf, mongo, queue_out):
    logger = logging.getLogger(__name__)
    mysql_settings = {'host': conf['host'], 'port': conf.getint('port'), 'user': conf['user'], 'passwd': conf['password']}
    res_dict = dict()
    dbs = conf['databases'].split(',')
    tables = ['already_table']
    for db in dbs:
        for table in tables:
            res_dict.update({table: mongo.get_log_pos(db, table)})
    for db in dbs:
        for table in tables:
            (log_file, log_pos, resume_stream) = res_dict.get(table)
            stream = BinLogStreamReader(connection_settings=mysql_settings, server_id=conf.getint('slaveid'), only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent], blocking=True, resume_stream=resume_stream, log_file=log_file, log_pos=log_pos, only_tables=table, only_schemas=db)
            for binlogevent in stream:
                binlogevent.dump()
                schema = ('%s' % binlogevent.schema)
                table = ('%s' % binlogevent.table)
                for row in binlogevent.rows:
                    if isinstance(binlogevent, DeleteRowsEvent):
                        vals = process_binlog_dict(row['values'])
                        event_type = 'delete'
                    elif isinstance(binlogevent, UpdateRowsEvent):
                        vals = dict()
                        vals['before'] = process_binlog_dict(row['before_values'])
                        vals['after'] = process_binlog_dict(row['after_values'])
                        event_type = 'update'
                    elif isinstance(binlogevent, WriteRowsEvent):
                        vals = process_binlog_dict(row['values'])
                        event_type = 'insert'
                    seqnum = mongo.write_to_queue(event_type, vals, schema, table)
                    mongo.write_log_pos(stream.log_file, stream.log_pos, db, table)
                    queue_out.put({'seqnum': seqnum})
                    logger.debug(f'------row------{row}')
                    logger.debug(f'------stream.log_pos------{stream.log_pos}')
                    logger.debug(f'------stream.log_file------{stream.log_file}')
    stream.close()
def mysql_stream(conf, mongo, queue_out):
    logger = logging.getLogger(__name__)
    mysql_settings = {'host': conf['host'], 'port': conf.getint('port'), 'user': conf['user'], 'passwd': conf['password']}
    res_dict = dict()
    dbs = conf['databases'].split(',')
    tables = ['already_table']
    for db in dbs:
        for table in tables:
            res_dict.update({table: mongo.get_log_pos(db, table)})
    for db in dbs:
        for table in tables:
            (log_file, log_pos, resume_stream) = res_dict.get(table)
            stream = BinLogStreamReader(connection_settings=mysql_settings, server_id=conf.getint('slaveid'), only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent], blocking=True, resume_stream=resume_stream, log_file=log_file, log_pos=log_pos, only_tables=table, only_schemas=db)
            for binlogevent in stream:
                binlogevent.dump()
                schema = ('%s' % binlogevent.schema)
                table = ('%s' % binlogevent.table)
                for row in binlogevent.rows:
                    if isinstance(binlogevent, DeleteRowsEvent):
                        vals = process_binlog_dict(row['values'])
                        event_type = 'delete'
                    elif isinstance(binlogevent, UpdateRowsEvent):
                        vals = dict()
                        vals['before'] = process_binlog_dict(row['before_values'])
                        vals['after'] = process_binlog_dict(row['after_values'])
                        event_type = 'update'
                    elif isinstance(binlogevent, WriteRowsEvent):
                        vals = process_binlog_dict(row['values'])
                        event_type = 'insert'
                    seqnum = mongo.write_to_queue(event_type, vals, schema, table)
                    mongo.write_log_pos(stream.log_file, stream.log_pos, db, table)
                    queue_out.put({'seqnum': seqnum})
                    logger.debug(f'------row------{row}')
                    logger.debug(f'------stream.log_pos------{stream.log_pos}')
                    logger.debug(f'------stream.log_file------{stream.log_file}')
    stream.close()