public void insertData(String schemaName, String tableName, DBObject naive, DBObject complete) {
    int i = 0;
    DBObject logObj = (DBObject) ObjectUtils.clone(complete);
    //保存原始数据
    try {
        String path = "/" + schemaName + "/" + tableName + "/" + CanalEntry.EventType.INSERT.getNumber();
        i++;
        naiveMongoTemplate.getCollection(tableName).insert(naive);
        i++;
        SpringUtil.doEvent(path, complete);
        i++;
    } catch (MongoClientException | MongoSocketException clientException) {
        //客户端连接异常抛出，阻塞同步，防止mongodb宕机
        throw clientException;
    } catch (Exception e) {
        logError(schemaName, tableName, 1, i, logObj, e);
    }
}
 

public void updateData(String schemaName, String tableName, DBObject query, DBObject obj) {
    String path = "/" + schemaName + "/" + tableName + "/" + CanalEntry.EventType.UPDATE.getNumber();
    int i = 0;
    DBObject newObj = (DBObject) ObjectUtils.clone(obj);
    DBObject logObj = (DBObject) ObjectUtils.clone(obj);
    //保存原始数据
    try {
        obj.removeField("id");
        i++;
        naiveMongoTemplate.getCollection(tableName).update(query, obj);
        i++;
        SpringUtil.doEvent(path, newObj);
        i++;
    } catch (MongoClientException | MongoSocketException clientException) {
        //客户端连接异常抛出，阻塞同步，防止mongodb宕机
        throw clientException;
    } catch (Exception e) {
        logError(schemaName, tableName, 2, i, logObj, e);
    }
}
 

public void deleteData(String schemaName, String tableName, DBObject obj) {
    int i = 0;
    String path = "/" + schemaName + "/" + tableName + "/" + CanalEntry.EventType.DELETE.getNumber();
    DBObject newObj = (DBObject) ObjectUtils.clone(obj);
    DBObject logObj = (DBObject) ObjectUtils.clone(obj);
    //保存原始数据
    try {
        i++;
        if (obj.containsField("id")) {
            naiveMongoTemplate.getCollection(tableName).remove(new BasicDBObject("_id", obj.get("id")));
        }
        i++;
        SpringUtil.doEvent(path, newObj);
    } catch (MongoClientException | MongoSocketException clientException) {
        //客户端连接异常抛出，阻塞同步，防止mongodb宕机
        throw clientException;
    } catch (Exception e) {
        logError(schemaName, tableName, 3, i, logObj, e);
    }
}
 

private void createIndexes() {
    try {
        Arrays.stream(schema.getJavaClass().getAnnotationsByType(MongoIndex.class))
                .forEach(this::createIndex);


        schema.getProperties().stream().filter(Property::isUnique)
                .forEach(property -> mongoCollection.createIndex(new BasicDBObject(property.getName(), 1)));
    } catch (MongoSocketException e) {
        logger.error("Failed at creating index", e);
    }

}
 

/**
 * Uses {@link ThreadLocalSpan} as there's no attribute namespace shared between callbacks, but
 * all callbacks happen on the same thread.
 */
@Override public void commandStarted(CommandStartedEvent event) {
  Span span = threadLocalSpan.next();
  if (span == null || span.isNoop()) return;

  String commandName = event.getCommandName();
  String databaseName = event.getDatabaseName();
  BsonDocument command = event.getCommand();
  String collectionName = getCollectionName(command, commandName);

  span.name(getSpanName(commandName, collectionName))
    .kind(CLIENT)
    .remoteServiceName("mongodb-" + databaseName)
    .tag("mongodb.command", commandName);

  if (collectionName != null) {
    span.tag("mongodb.collection", collectionName);
  }

  ConnectionDescription connectionDescription = event.getConnectionDescription();
  if (connectionDescription != null) {
    ConnectionId connectionId = connectionDescription.getConnectionId();
    if (connectionId != null) {
      span.tag("mongodb.cluster_id", connectionId.getServerId().getClusterId().getValue());
    }

    try {
      InetSocketAddress socketAddress =
        connectionDescription.getServerAddress().getSocketAddress();
      span.remoteIpAndPort(socketAddress.getAddress().getHostAddress(), socketAddress.getPort());
    } catch (MongoSocketException ignored) {

    }
  }

  span.start();
}