import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
void dispatch(Event evt) {
  Object data = evt.getData();
  if (TableMapEventData.class.isInstance(data)) {
    handleTableMapEvent((TableMapEventData) data);
  } else if (EventType.isWrite(evt.getHeader().getEventType())) {
    handleWriteEvent((WriteRowsEventData) data);
  } else if (EventType.isUpdate(evt.getHeader().getEventType())) {
    handleUpdateEvent((UpdateRowsEventData) data);
  } else if (EventType.isDelete(evt.getHeader().getEventType())) {
    handleDeleteEvent((DeleteRowsEventData) data);
  } else if (QueryEventData.class.isInstance(data)) {
    handleQueryEvent((QueryEventData) data);
  }
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
private void handleDeleteRows(DeleteRowsEventData data, List<DataChange> dataChanges) throws JsonProcessingException {
    System.out.println(data);
    TableMapEventData tableData = tableIdToDataMap.get(data.getTableId());
    Pair<String, String> tableUniqueName = new ImmutablePair<String, String>(tableData.getDatabase(), 
            tableData.getTable());
    
    TableMetadata tableMetadata = databaseNameToTableMetadata.get(tableUniqueName);
    // XXX Error handling if tableMetadata == null
    ArrayList<String> columnNames = tableMetadata.getColumnNames();
    List<Integer> primaryKeyIndices = tableMetadata.getPrimaryKeyIndices();
    
    List<Serializable[]> rows = data.getRows();
    for (Serializable[] row: rows) {
        
        Map<String, Serializable> primaryKeyMap = new HashMap<String, Serializable>();
        for (int index : primaryKeyIndices) {
            String columnName = columnNames.get(index);
            Serializable value = row[index];
            primaryKeyMap.put(columnName, value);
        }
       
        DataChange change = new DataChange();
        change.setPayload(null);
        change.setPrimaryKey(primaryKeyMap);
        change.setTableName(tableUniqueName);
        
        dataChanges.add(change);
    }
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
private void handleWriteRows(WriteRowsEventData data, List<DataChange> dataChanges) throws JsonProcessingException {
        TableMapEventData tableData = tableIdToDataMap.get(data.getTableId());
        Pair<String, String> tableUniqueName = new ImmutablePair<String, String>(tableData.getDatabase(), 
                tableData.getTable());
        TableMetadata tableMetadata = databaseNameToTableMetadata.get(tableUniqueName);
        // XXX Error handling if tableMetadata == null

        ArrayList<String> columnNames = tableMetadata.getColumnNames();
        List<Integer> primaryKeyIndices = tableMetadata.getPrimaryKeyIndices();
        
        List<Serializable[]> rows = data.getRows();
        for (Serializable[] row: rows) {
            Map<String,Serializable> map = new HashMap<String, Serializable>();

//            System.out.println(row);
            for (int i = 0; i < row.length; i++) {
                String column = columnNames.get(i);
                map.put(column, row[i]);
            }

            Map<String, Serializable> primaryKeyMap = new HashMap<String, Serializable>();
            for (int index : primaryKeyIndices) {
                String columnName = columnNames.get(index);
                Serializable value = row[index];
                primaryKeyMap.put(columnName, value);
            }

            DataChange change = new DataChange();
            change.setPayload(map);
            change.setPrimaryKey(primaryKeyMap);
            change.setTableName(tableUniqueName);
            
            dataChanges.add(change);
        }
    }
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
public Event createTableMapEvent(int tableId, String database, String table) {
	EventHeaderV4 eventHeader = new EventHeaderV4();
	eventHeader.setEventType(EventType.TABLE_MAP);
	
	TableMapEventData eventData = new TableMapEventData();
	eventData.setTableId(tableId);
	eventData.setDatabase(database);
	eventData.setTable(table);
	
	return new Event(eventHeader, eventData);
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
public long getTableId() {
    if (this.USING_DEPRECATED_PARSER) {
        return  ((TableMapEvent) this.getBinlogEventV4()).getTableId();
    }
    else {
        return ((TableMapEventData) this.getBinlogConnectorEvent().getData()).getTableId();
    }
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
public String getTableName() {
    if (this.USING_DEPRECATED_PARSER) {
        return  ((TableMapEvent) this.getBinlogEventV4()).getTableName().toString();
    }
    else {
        return ((TableMapEventData) this.getBinlogConnectorEvent().getData()).getTable();
    }
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
public void setTableName(String tableName) {
    if (this.USING_DEPRECATED_PARSER) {
        ((TableMapEvent) this.getBinlogEventV4()).setTableName(
                StringColumn.valueOf(tableName.getBytes())
        );
    }
    else {
        ((TableMapEventData) this.getBinlogConnectorEvent().getData()).setTable(tableName);
    }
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
public String getDatabaseName() {
    if (this.USING_DEPRECATED_PARSER) {
        return  ((TableMapEvent) this.getBinlogEventV4()).getDatabaseName().toString();
    }
    else {
        return ((TableMapEventData) this.getBinlogConnectorEvent().getData()).getDatabase();
    }
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
public void setDatabaseName(String databaseName) {
    if (this.USING_DEPRECATED_PARSER) {
        ((TableMapEvent) this.getBinlogEventV4()).setDatabaseName(
                StringColumn.valueOf(databaseName.getBytes())
        );
    }
    else {
        ((TableMapEventData) this.getBinlogConnectorEvent().getData()).setDatabase(databaseName);
    }
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
public DeleteEvent(TableMapEventData tableMapEvent,
                   List<Serializable[]> rows,
                   byte[] usedColumns) {

    this.tableMapEvent = tableMapEvent;
    this.rows = rows;
    this.usedColumns = usedColumns;
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
public WriteEvent(TableMapEventData tableMapEvent,
                  List<Serializable[]> rows,
                  byte[] usedColumns) {

    this.tableMapEvent = tableMapEvent;
    this.rows = rows;
    this.usedColumns = usedColumns;
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
public UpdateEvent(TableMapEventData tableMapEvent,
                   List<Map.Entry<Serializable[],Serializable[]>> rows,
                   byte[] usedColumnsBefore,
                   byte[] usedColumnsAfter) {

    this.tableMapEvent = tableMapEvent;
    this.rows = rows;
    this.usedColumnsBefore = usedColumnsBefore;
    this.usedColumnsAfter = usedColumnsAfter;
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
private void handleTableMapEvent(TableMapEventData data) {
  lastTableMap = data;
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
@Override
public void onEvent(Event event) {
    try {

        EventData data = event.getData();

        if (! (event.getHeader() instanceof EventHeaderV4)) {
            throw new RuntimeException("Header is not V4!");
        }
        System.out.println(event);  

        EventHeaderV4 header = event.getHeader();

        List<DataChange> dataChanges = new ArrayList<DataChange>();
        
        EventType type = header.getEventType();
        if (type == EventType.TABLE_MAP) {
            handleTableMap((TableMapEventData) data);
        } else if (EventType.isWrite(type)) {
            handleWriteRows((WriteRowsEventData) data, dataChanges);
        } else if (EventType.isDelete(type)) {
            handleDeleteRows((DeleteRowsEventData) data, dataChanges);
        } else if (EventType.isUpdate(type)) {
            handleUpdateRows((UpdateRowsEventData) data, dataChanges);
        } else if (type == EventType.QUERY) {
            handleQuery((QueryEventData) data, header);
        } else if (type == EventType.ROTATE) {
            handleRotate((RotateEventData) data);
        } else if (type == EventType.XID) {
            handleXid((XidEventData) data, header);
        }

        for (DataChange change : dataChanges) {
            change.setOffset(this.transactionStartPosition,
                    this.binlogFilename, header.getPosition());
            change.setServerId(this.mysqlServerId);
            outputter.output(change);
        }
        
        outputter.flush();
        
    } catch (Throwable e) {
        // during development, on any and all errors, disconnect.
        // This allows me to learn what errors will happen, and how to
        // handle each one appropriately.
        e.printStackTrace();
        try {
            this.binlogClient.disconnect();
        } catch (IOException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }

    }
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
private void handleTableMap(TableMapEventData data) throws SQLException {
    Pair<String, String> tableUniqueName = new ImmutablePair<String, String>(data.getDatabase(), data.getTable());
    if (databaseNameToTableIdMap.containsKey(tableUniqueName)) {
        // delete the id->info mapping
        System.out.println("Removing old db info!");
        long oldTableId = databaseNameToTableIdMap.get(tableUniqueName);
        tableIdToDataMap.remove(oldTableId);
    }
    tableIdToDataMap.put(data.getTableId(), data);
    databaseNameToTableIdMap.put(tableUniqueName, data.getTableId());

    if (databaseNameToTableMetadata.containsKey(tableUniqueName)) {
        return;
    }
    
    // lookup the data
    TableMetadata tableMetadata = new TableMetadata();

    String dbName = data.getDatabase();
    String tableName = data.getTable();

    // XXX I should close this connection!
    Connection connection;
    connection = DriverManager.getConnection(
            "jdbc:mysql://" + hostname + ":" + port + "/" + dbName, // notice the dbname is here. mysql's jdbc driver requires it. 
            username,
            password);
    DatabaseMetaData databaseMetaData = connection.getMetaData();

    // get column names
    ResultSet rs = databaseMetaData.getColumns(null, null, tableName, null);
    // what about failures to connect to database? Need to make sure that we catch the case where we add an empty list, and then fail (meaning, no column names)
    ArrayList<String> columnNames = new ArrayList<String>();

    while (rs.next()) {
        String columnName = rs.getString("COLUMN_NAME");
        // 1 based, but we don't care. We just add them in order. This
        // assumes that the columns come in ORDINAL_POSITION order, but the
        // docs for getColumns() says they do.
        // String position = rs.getString("ORDINAL_POSITION");
        columnNames.add(columnName);
    }
    tableMetadata.setColumnNames(columnNames);

    // get primary keys
    ResultSet rs2 = databaseMetaData.getPrimaryKeys(null, null, tableName);

    List<String> primaryKeyColumns = new ArrayList<String>();
    while (rs2.next()) {
        primaryKeyColumns.add(rs2.getString("COLUMN_NAME"));
    }
    if (primaryKeyColumns.isEmpty()) {
        throw new RuntimeException(String.format("Table %s.%s does not have a primary key!", data.getDatabase(), data.getTable()));
    }
    tableMetadata.setPrimaryKeys(primaryKeyColumns);
    System.out.println(tableMetadata);

    // Only upon successful lookup, with no sql errors, do we save the data.
    databaseNameToTableMetadata.put(tableUniqueName, tableMetadata);
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
@Override
public void onEvent(Event event) {
  LOG.trace("Received event {}", event);
  EventType eventType = event.getHeader().getEventType();
  currentBinLogFileName = client.getBinlogFilename();
  switch (eventType) {
    case TABLE_MAP:
      handleTableMappingEvent((TableMapEventData) event.getData());
      break;
    case PRE_GA_WRITE_ROWS:
    case WRITE_ROWS:
    case EXT_WRITE_ROWS:
      handleRowEvent(event, event.<WriteRowsEventData>getData().getTableId());
      break;
    case PRE_GA_UPDATE_ROWS:
    case UPDATE_ROWS:
    case EXT_UPDATE_ROWS:
      handleRowEvent(event, event.<UpdateRowsEventData>getData().getTableId());
      break;
    case PRE_GA_DELETE_ROWS:
    case DELETE_ROWS:
    case EXT_DELETE_ROWS:
      handleRowEvent(event, event.<DeleteRowsEventData>getData().getTableId());
      break;
    case QUERY:
      QueryEventData queryEventData = event.getData();
      String query = queryEventData.getSql();
      if (isCommit(query)) {
        finishTx();
      } else if (isSchemaChangeQuery(query)) {
        schemaRepository.evictAll();
      }
      break;
    case XID:
      finishTx();
      break;
    case GTID:
      GtidEventData eventData = event.getData();
      currentGtidSet = client.getGtidSet();
      currentTxGtid = eventData.getGtid();
      currentTxEventSeqNo = 0;
      LOG.trace("Started new tx, gtid: {}", currentTxGtid);
      break;
    default:
      // ignore
      break;
  }
}
 

import com.github.shyiko.mysql.binlog.event.TableMapEventData; //导入依赖的package包/类
private void handleTableMappingEvent(TableMapEventData eventData) {
  tableMapping.put(
      eventData.getTableId(),
      new DatabaseAndTable(eventData.getDatabase(), eventData.getTable())
  );
}