聊聊SpinalTap的BinlogEventListener

@Slf4j
public final class BinaryLogConnectorSource extends MysqlSource {

  //......

  private final class BinlogEventListener implements BinaryLogClient.EventListener {
    public void onEvent(Event event) {
      Preconditions.checkState(isStarted(), "Source is not started and should not process events");

      final EventHeaderV4 header = event.getHeader();
      final BinlogFilePos filePos =
          new BinlogFilePos(
              binlogClient.getBinlogFilename(),
              header.getPosition(),
              header.getNextPosition(),
              binlogClient.getGtidSet(),
              serverUUID);

      BinaryLogConnectorEventMapper.INSTANCE
          .map(event, filePos)
          .ifPresent(BinaryLogConnectorSource.super::processEvent);
    }
  }

  //......

}

@Slf4j
@Getter
@EqualsAndHashCode
@NoArgsConstructor
@JsonDeserialize(builder = BinlogFilePos.Builder.class)
public class BinlogFilePos implements Comparable<BinlogFilePos>, Serializable {
  private static final long serialVersionUID = 1549638989059430876L;

  private static final Splitter SPLITTER = Splitter.on(':');
  private static final String NULL_VALUE = "null";
  public static final String DEFAULT_BINLOG_FILE_NAME = "mysql-bin-changelog";

  @JsonProperty private String fileName;
  @JsonProperty private long position;
  @JsonProperty private long nextPosition;
  @JsonProperty private GtidSet gtidSet;
  @JsonProperty private String serverUUID;

  public BinlogFilePos(long fileNumber) {
    this(fileNumber, 4L, 4L);
  }

  public BinlogFilePos(String fileName) {
    this(fileName, 4L, 4L);
  }

  public BinlogFilePos(long fileNumber, long position, long nextPosition) {
    this(String.format("%s.%06d", DEFAULT_BINLOG_FILE_NAME, fileNumber), position, nextPosition);
  }

  public BinlogFilePos(
      String fileName, long position, long nextPosition, String gtidSet, String serverUUID) {
    this.fileName = fileName;
    this.position = position;
    this.nextPosition = nextPosition;
    this.serverUUID = serverUUID;
    if (gtidSet != null) {
      this.gtidSet = new GtidSet(gtidSet);
    }
  }

  public BinlogFilePos(String fileName, long position, long nextPosition) {
    this(fileName, position, nextPosition, null, null);
  }

  public static BinlogFilePos fromString(@NonNull final String position) {
    Iterator<String> parts = SPLITTER.split(position).iterator();
    String fileName = parts.next();
    String pos = parts.next();
    String nextPos = parts.next();

    if (NULL_VALUE.equals(fileName)) {
      fileName = null;
    }

    return new BinlogFilePos(fileName, Long.parseLong(pos), Long.parseLong(nextPos));
  }

  @JsonIgnore
  public long getFileNumber() {
    if (fileName == null) {
      return Long.MAX_VALUE;
    }
    if (fileName.equals("")) {
      return Long.MIN_VALUE;
    }
    String num = fileName.substring(fileName.lastIndexOf('.') + 1);
    return Long.parseLong(num);
  }

  @Override
  public String toString() {
    return String.format("%s:%d:%d", fileName, position, nextPosition);
  }

  @Override
  public int compareTo(@NonNull final BinlogFilePos other) {
    if (shouldCompareUsingFilePosition(this, other)) {
      return getFileNumber() != other.getFileNumber()
          ? Long.compare(getFileNumber(), other.getFileNumber())
          : Long.compare(getPosition(), other.getPosition());
    }

    if (this.gtidSet.equals(other.gtidSet)) {
      return 0;
    }
    if (this.gtidSet.isContainedWithin(other.gtidSet)) {
      return -1;
    }
    return 1;
  }

  /** Check if two BinlogFilePos are from the same source MySQL server */
  private static boolean isFromSameSource(BinlogFilePos pos1, BinlogFilePos pos2) {
    return pos1.getServerUUID() != null
        && pos1.getServerUUID().equalsIgnoreCase(pos2.getServerUUID());
  }

  /** Whether we can compare two BinlogFilePos using Binlog file position (without GTIDSet) */
  public static boolean shouldCompareUsingFilePosition(BinlogFilePos pos1, BinlogFilePos pos2) {
    return isFromSameSource(pos1, pos2) || pos1.getGtidSet() == null || pos2.getGtidSet() == null;
  }

  //......

}

@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class BinaryLogConnectorEventMapper {
  public static final BinaryLogConnectorEventMapper INSTANCE = new BinaryLogConnectorEventMapper();

  public Optional<BinlogEvent> map(
      @NonNull final Event event, @NonNull final BinlogFilePos position) {
    final EventHeaderV4 header = event.getHeader();
    final EventType eventType = header.getEventType();
    final long serverId = header.getServerId();
    final long timestamp = header.getTimestamp();

    if (EventType.isWrite(eventType)) {
      final WriteRowsEventData data = event.getData();
      return Optional.of(
          new WriteEvent(data.getTableId(), serverId, timestamp, position, data.getRows()));
    } else if (EventType.isUpdate(eventType)) {
      final UpdateRowsEventData data = event.getData();
      return Optional.of(
          new UpdateEvent(data.getTableId(), serverId, timestamp, position, data.getRows()));
    } else if (EventType.isDelete(eventType)) {
      final DeleteRowsEventData data = event.getData();
      return Optional.of(
          new DeleteEvent(data.getTableId(), serverId, timestamp, position, data.getRows()));
    } else {
      switch (eventType) {
        case TABLE_MAP:
          TableMapEventData tableMapData = event.getData();
          return Optional.of(
              new TableMapEvent(
                  tableMapData.getTableId(),
                  serverId,
                  timestamp,
                  position,
                  tableMapData.getDatabase(),
                  tableMapData.getTable(),
                  tableMapData.getColumnTypes()));
        case XID:
          final XidEventData xidData = event.getData();
          return Optional.of(new XidEvent(serverId, timestamp, position, xidData.getXid()));
        case GTID:
          final GtidEventData gtidEventData = event.getData();
          return Optional.of(new GTIDEvent(serverId, timestamp, position, gtidEventData.getGtid()));
        case QUERY:
          final QueryEventData queryData = event.getData();
          return Optional.of(
              new QueryEvent(
                  serverId,
                  timestamp,
                  position,
                  queryData.getDatabase(),
                  queryData
                      .getSql()
                      // https://dev.mysql.com/doc/refman/5.7/en/comments.html
                      // Replace MySQL-specific comments (/*! ... */ and /*!50110 ... */) which
                      // are actually executed
                      .replaceAll("/\\*!(?:\\d{5})?(.*?)\\*/", "$1")
                      // Remove block comments
                      // https://stackoverflow.com/questions/13014947/regex-to-match-a-c-style-multiline-comment
                      // line comments and newlines are kept
                      // Note: This does not handle comments in quotes
                      .replaceAll("/\\*[^*]*\\*+(?:[^/*][^*]*\\*+)*/", " ")
                      // Remove extra spaces
                      .replaceAll("\\h+", " ")
                      .replaceAll("^\\s+", "")));
        case FORMAT_DESCRIPTION:
          return Optional.of(new StartEvent(serverId, timestamp, position));
        default:
          return Optional.empty();
      }
    }
  }
}

@Slf4j
@RequiredArgsConstructor
public abstract class AbstractSource<E extends SourceEvent> extends ListenableSource<E> {

  //......

  public final void processEvent(final E event) {
    try {
      if (!eventFilter.apply(event)) {
        log.debug("Event filtered from source {}. Skipping. event={}", name, event);
        return;
      }

      notifyEvent(event);

      final Stopwatch stopwatch = Stopwatch.createStarted();

      metrics.eventReceived(event);
      log.debug("Received event from source {}. event={}", name, event);

      notifyMutations(mutationMapper.map(event));

      stopwatch.stop();
      final long time = stopwatch.elapsed(TimeUnit.MILLISECONDS);

      metrics.processEventTime(event, time);

    } catch (Exception ex) {
      if (!isStarted()) {
        // Do not process the exception if streaming has stopped.
        return;
      }

      final String errorMessage = String.format("Failed to process event from source %s", name);

      log.error(errorMessage, ex);
      metrics.eventFailure(ex);

      notifyError(ex);

      throw new SourceException(errorMessage, ex);
    }
  }

  //......

}

abstract class ListenableSource<E extends SourceEvent> implements Source {
  private final List<Listener> listeners = new ArrayList<>();

  @Override
  public void addListener(@NonNull final Listener listener) {
    listeners.add(listener);
  }

  @Override
  public void removeListener(@NonNull final Listener listener) {
    listeners.remove(listener);
  }

  protected void notifyMutations(final List<? extends Mutation<?>> mutations) {
    if (!mutations.isEmpty()) {
      listeners.forEach(listener -> listener.onMutation(mutations));
    }
  }

  protected void notifyEvent(E event) {
    listeners.forEach(listener -> listener.onEvent(event));
  }

  protected void notifyError(Throwable error) {
    listeners.forEach(listener -> listener.onError(error));
  }

  protected void notifyStart() {
    listeners.forEach(Source.Listener::onStart);
  }
}

  final class SourceListener extends Source.Listener {
    public void onMutation(List<? extends Mutation<?>> mutations) {
      destination.send(mutations);
    }

    public void onError(Throwable error) {
      errorHandlingExecutor.execute(Pipe.this::close);
    }
  }