import org.onosproject.ovsdb.rfc.message.TableUpdates; //导入依赖的package包/类
@Override
public ListenableFuture<TableUpdates> monitorTables(String dbName, String id) {
    if (dbName == null) {
        return null;
    }
    DatabaseSchema dbSchema = schema.get(dbName);
    if (dbSchema != null) {
        Function<JsonNode, TableUpdates> rowFunction = new Function<JsonNode, TableUpdates>() {
            @Override
            public TableUpdates apply(JsonNode input) {
                log.info("Get table updates");
                TableUpdates updates = FromJsonUtil
                        .jsonNodeToTableUpdates(input, dbSchema);
                if (updates == null) {
                    log.debug("Get table updates error");
                    return null;
                }
                return updates;
            }
        };
        return Futures.transform(monitor(dbSchema, id), rowFunction);
    }
    return null;
}
 

import org.onosproject.ovsdb.rfc.message.TableUpdates; //导入依赖的package包/类
@Override
public void update(UpdateNotification updateNotification) {
    Object key = updateNotification.jsonValue();
    OvsdbClientService ovsdbClient = requestNotification.get(key);

    String dbName = requestDbName.get(key);
    JsonNode updatesJson = updateNotification.tbUpdatesJsonNode();
    DatabaseSchema dbSchema = ovsdbClient.getDatabaseSchema(dbName);
    TableUpdates updates = FromJsonUtil
            .jsonNodeToTableUpdates(updatesJson, dbSchema);
    try {
        processTableUpdates(ovsdbClient, updates, dbName);
    } catch (InterruptedException e) {
        log.warn("Interrupted while processing table updates");
        Thread.currentThread().interrupt();
    }
}
 

import org.onosproject.ovsdb.rfc.message.TableUpdates; //导入依赖的package包/类
@Override
public ListenableFuture<TableUpdates> monitorTables(String dbName, String id) {
    if (dbName == null) {
        return null;
    }
    DatabaseSchema dbSchema = schema.get(dbName);
    if (dbSchema != null) {
        Function<JsonNode, TableUpdates> rowFunction = input -> {
            log.debug("Get table updates");
            TableUpdates updates = FromJsonUtil.jsonNodeToTableUpdates(input, dbSchema);
            if (updates == null) {
                log.debug("Get table updates error");
                return null;
            }
            return updates;
        };
        return Futures.transform(monitor(dbSchema, id), rowFunction);
    }
    return null;
}
 

import org.onosproject.ovsdb.rfc.message.TableUpdates; //导入依赖的package包/类
/**
 * Processes table updates.
 *
 * @param clientService OvsdbClientService instance
 * @param updates       TableUpdates instance
 * @param dbName        ovsdb database name
 */
private void processTableUpdates(OvsdbClientService clientService,
                                 TableUpdates updates, String dbName)
        throws InterruptedException {
    checkNotNull(clientService, "OvsdbClientService is not null");

    DatabaseSchema dbSchema = clientService.getDatabaseSchema(dbName);

    for (String tableName : updates.result().keySet()) {
        TableUpdate update = updates.result().get(tableName);
        for (Uuid uuid : (Set<Uuid>) update.rows().keySet()) {
            log.debug("Begin to process table updates uuid: {}, databaseName: {}, tableName: {}",
                      uuid.value(), dbName, tableName);

            Row newRow = update.getNew(uuid);
            if (newRow != null) {
                clientService.updateOvsdbStore(dbName, tableName,
                                               uuid.value(), newRow);

                if (OvsdbConstant.INTERFACE.equals(tableName)) {
                    dispatchInterfaceEvent(clientService,
                                           newRow,
                                           OvsdbEvent.Type.PORT_ADDED,
                                           dbSchema);
                }
            } else if (update.getOld(uuid) != null) {
                if (OvsdbConstant.INTERFACE.equals(tableName)) {
                    Row row = clientService.getRow(OvsdbConstant.DATABASENAME, tableName, uuid.value());
                    dispatchInterfaceEvent(clientService,
                                           row,
                                           OvsdbEvent.Type.PORT_REMOVED,
                                           dbSchema);
                }
                clientService.removeRow(dbName, tableName, uuid.value());
            }
        }
    }
}
 

import org.onosproject.ovsdb.rfc.message.TableUpdates; //导入依赖的package包/类
/**
 * convert the params of Update Notification into TableUpdates.
 * @param updatesJson the params of Update Notification
 * @param dbSchema DatabaseSchema entity
 * @return TableUpdates
 */
public static TableUpdates jsonNodeToTableUpdates(JsonNode updatesJson, DatabaseSchema dbSchema) {
    Map<String, TableUpdate> tableUpdateMap = Maps.newHashMap();
    Iterator<Map.Entry<String, JsonNode>> tableUpdatesItr = updatesJson.fields();
    while (tableUpdatesItr.hasNext()) {
        Map.Entry<String, JsonNode> entry = tableUpdatesItr.next();
        TableSchema tableSchema = dbSchema.getTableSchema(entry.getKey());
        TableUpdate tableUpdate = jsonNodeToTableUpdate(tableSchema, entry.getValue());
        tableUpdateMap.put(entry.getKey(), tableUpdate);
    }
    return TableUpdates.tableUpdates(tableUpdateMap);
}
 

import org.onosproject.ovsdb.rfc.message.TableUpdates; //导入依赖的package包/类
@Override
public ListenableFuture<TableUpdates> monitorTables(String dbName, String id) {
    return null;
}
 

import org.onosproject.ovsdb.rfc.message.TableUpdates; //导入依赖的package包/类
/**
 * Gets the OVSDB table updates.
 *
 * @param dbName database name
 * @param id     random uuid
 * @return table updates
 */
ListenableFuture<TableUpdates> monitorTables(String dbName, String id);