import org.onlab.util.Tools; //导入依赖的package包/类
@Activate
protected void activate() {
    cache = new ModelCache(new DefaultServiceBundle(), eventDispatcher);
    eventHandler = Executors.newSingleThreadExecutor(Tools.groupedThreads("onos/ui/topo", "event-handler"));

    eventDispatcher.addSink(UiModelEvent.class, listenerRegistry);

    clusterService.addListener(clusterListener);
    mastershipService.addListener(mastershipListener);
    regionService.addListener(regionListener);
    deviceService.addListener(deviceListener);
    linkService.addListener(linkListener);
    hostService.addListener(hostListener);
    intentService.addListener(intentListener);
    flowService.addListener(flowRuleListener);

    cache.load();

    log.info("Started");
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
/**
 * Displays text representing the leaders.
 *
 * @param leaderBoard map of leaders
 */
private void displayLeaders(Map<String, Leadership> leaderBoard) {
    print("------------------------------------------------------------------------");
    print(FMT, "Topic", "Leader", "Term", "Elected");
    print("------------------------------------------------------------------------");

    leaderBoard.values()
            .stream()
            .filter(l -> allTopics || pattern.matcher(l.topic()).matches())
            .filter(l -> l.leader() != null)
            .sorted(leadershipComparator)
            .forEach(l -> print(FMT,
                    l.topic(),
                    l.leaderNodeId(),
                    l.leader().term(),
                    Tools.timeAgo(l.leader().termStartTime())));
    print("------------------------------------------------------------------------");
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
protected void execute() {
    ClusterAdminService service = get(ClusterAdminService.class);
    List<ControllerNode> nodes = newArrayList(service.getNodes());
    Collections.sort(nodes, Comparators.NODE_COMPARATOR);
    if (outputJson()) {
        print("%s", json(service, nodes));
    } else {
        ControllerNode self = service.getLocalNode();
        for (ControllerNode node : nodes) {
            DateTime lastUpdated = service.getLastUpdated(node.id());
            String timeAgo = "Never";
            if (lastUpdated != null) {
                timeAgo = Tools.timeAgo(lastUpdated.getMillis());
            }
            print(FMT, node.id(), node.ip(), node.tcpPort(),
                  service.getState(node.id()), timeAgo,
                  node.equals(self) ? "*" : "");
        }
    }
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Modified
protected void modified(ComponentContext context) {
    Dictionary<?, ?> properties = context.getProperties();
    String updatedUrl;

    updatedUrl = Tools.get(properties, VTN_SERVICE_URL);
    if (!Strings.isNullOrEmpty(updatedUrl) && !updatedUrl.equals(vtnServiceUrl)) {
        vtnServiceUrl = updatedUrl;
        vtnServiceApi = new DefaultVtnServiceApi(vtnServiceUrl, access);
    }

   updatedUrl = Tools.get(properties, VTN_PORT_URL);
    if (!Strings.isNullOrEmpty(updatedUrl) && !updatedUrl.equals(vtnPortUrl)) {
        vtnPortUrl = updatedUrl;
        vtnPortApi = new DefaultVtnPortApi(vtnPortUrl, access);
    }

    log.info("Modified");
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
/**
 * Extracts properties from the component configuration context.
 *
 * @param context the component context
 */
private void readComponentConfiguration(ComponentContext context) {
    Dictionary<?, ?> properties = context.getProperties();

    String newXosServerAddress =
            Tools.get(properties, XOS_SERVER_ADDRESS_PROPERTY_NAME);
    if (!isNullOrEmpty(newXosServerAddress)) {
        xosServerAddress = newXosServerAddress;
    }

    String newXosServerPortString =
            Tools.get(properties, XOS_SERVER_PORT_PROPERTY_NAME);
    if (!isNullOrEmpty(newXosServerPortString)) {
        xosServerPort = Integer.parseInt(newXosServerPortString);
    }

    String newXosProviderServiceString =
            Tools.get(properties, XOS_PROVIDER_SERVICE_PROPERTY_NAME);
    if (!isNullOrEmpty(newXosProviderServiceString)) {
        xosProviderService = Integer.parseInt(newXosProviderServiceString);
    }
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
/**
 * Extracts properties from the component configuration context.
 *
 * @param context the component context
 */
private void readComponentConfiguration(ComponentContext context) {
    Dictionary<?, ?> properties = context.getProperties();

    String metricNameStr = Tools.get(properties, "metricNames");
    metricNames = metricNameStr != null ? metricNameStr : DEFAULT_METRIC_NAMES;
    log.info("Configured. Metric name is {}", metricNames);

    Boolean monitorAllEnabled = Tools.isPropertyEnabled(properties, "monitorAll");
    if (monitorAllEnabled == null) {
        log.info("Monitor all metrics is not configured, " +
                "using current value of {}", monitorAll);
    } else {
        monitorAll = monitorAllEnabled;
        log.info("Configured. Monitor all metrics is {}",
                monitorAll ? "enabled" : "disabled");
    }
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
protected void execute() {
    PimInterfaceService interfaceService = get(PimInterfaceService.class);

    Set<PimInterface> interfaces = interfaceService.getPimInterfaces();

    for (PimInterface intf : interfaces) {
        print(INTF_FORMAT, intf.getInterface().name(), intf.getIpAddress());
        for (PimNeighbor neighbor : intf.getNeighbors()) {
            // Filter out the PIM neighbor representing 'us'
            if (!neighbor.ipAddress().equals(intf.getIpAddress())) {
                print(NEIGHBOR_FORMAT, neighbor.ipAddress(),
                        Tools.timeAgo(neighbor.upTime()), neighbor.holdtime(),
                        neighbor.priority(), neighbor.generationId());
            }
        }
    }
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
public void formCluster(Set<ControllerNode> nodes) {
    checkNotNull(nodes, "Nodes cannot be null");
    checkArgument(!nodes.isEmpty(), "Nodes cannot be empty");

    ClusterMetadata metadata = new ClusterMetadata("default", nodes, buildDefaultPartitions(nodes));
    clusterMetadataAdminService.setClusterMetadata(metadata);
    try {
        log.warn("Shutting down container for cluster reconfiguration!");
        // Clean up persistent state associated with previous cluster configuration.
        Tools.removeDirectory(System.getProperty("karaf.data") + "/partitions");
        systemService.reboot("now", SystemService.Swipe.NONE);
    } catch (Exception e) {
        log.error("Unable to reboot container", e);
    }
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Modified
public void modified(ComponentContext context) {
    if (context == null) {
        log.info("Settings: useFlowObjectives={}", useFlowObjectives);
        return;
    }

    boolean newFlowObjectives;
    try {
        String s = Tools.get(context.getProperties(), "useFlowObjectives");
        newFlowObjectives = isNullOrEmpty(s) ? useFlowObjectives : Boolean.parseBoolean(s.trim());
    } catch (ClassCastException e) {
        newFlowObjectives = useFlowObjectives;
    }

    if (useFlowObjectives != newFlowObjectives) {
        useFlowObjectives = newFlowObjectives;
        changeCompilers();
        log.info("Settings: useFlowObjectives={}", useFlowObjectives);
    }
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
/**
 * Returns existing and available optical connectivity intent that matches the given circuit intent.
 *
 * @param src source connect point of optical circuit intent
 * @param dst destination connect point of optical circuit intent
 * @param signalType signal type of optical circuit intent
 * @param multiplexingSupported indicates whether ODU multiplexing is supported
 * @return existing optical connectivity intent, null otherwise.
 */
private OpticalConnectivityIntent findOpticalConnectivityIntent(ConnectPoint src,
                                                                ConnectPoint dst,
                                                                CltSignalType signalType,
                                                                boolean multiplexingSupported) {

    OduSignalType oduSignalType = OduSignalUtils.mappingCltSignalTypeToOduSignalType(signalType);

    return Tools.stream(intentService.getIntents())
            .filter(x -> x instanceof OpticalConnectivityIntent)
            .map(x -> (OpticalConnectivityIntent) x)
            .filter(x -> src.deviceId().equals(x.getSrc().deviceId()))
            .filter(x -> dst.deviceId().equals(x.getDst().deviceId()))
            .filter(x -> isAllowed(src, x.getSrc()))
            .filter(x -> isAllowed(dst, x.getDst()))
            .filter(x -> isAvailable(x.id()))
            .filter(x -> !multiplexingSupported ||
                    isAvailableTributarySlots(x.getSrc(), x.getDst(), oduSignalType.tributarySlots()))
            .findFirst()
            .orElse(null);
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
/**
 * Extracts properties from the component configuration context.
 *
 * @param context the component context
 */
private void readComponentConfiguration(ComponentContext context) {
    Dictionary<?, ?> properties = context.getProperties();

    Integer newPendingFutureTimeoutMinutes =
            Tools.getIntegerProperty(properties, "pendingFutureTimeoutMinutes");
    if (newPendingFutureTimeoutMinutes == null) {
        pendingFutureTimeoutMinutes = DEFAULT_PENDING_FUTURE_TIMEOUT_MINUTES;
        log.info("Pending future timeout is not configured, " +
                         "using current value of {}", pendingFutureTimeoutMinutes);
    } else {
        pendingFutureTimeoutMinutes = newPendingFutureTimeoutMinutes;
        log.info("Configured. Pending future timeout is configured to {}",
                 pendingFutureTimeoutMinutes);
    }
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
public CompletableFuture<Void> addListener(AtomicValueEventListener<String> listener) {
    // TODO: synchronization
    if (mapEventListener == null) {
        mapEventListener = event -> {
            Versioned<byte[]> newValue = event.newValue();
            Versioned<byte[]> oldValue = event.oldValue();
            if (Objects.equals(event.key(), name)) {
                listener.event(new AtomicValueEvent<>(name,
                        newValue == null ? null : Tools.toStringUtf8(newValue.value()),
                        oldValue == null ? null : Tools.toStringUtf8(oldValue.value())));
            }
        };
        return atomixMap.addListener(mapEventListener).whenComplete((r, e) -> {
            if (e == null) {
                listeners.add(listener);
            } else {
                mapEventListener = null;
            }
        });
    } else {
        listeners.add(listener);
        return CompletableFuture.completedFuture(null);
    }
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
public CompletableFuture<Boolean> prepare(MapTransaction<K, V> transaction) {

    Map<AsyncConsistentMap<K, V>, List<MapUpdate<K, V>>> updatesGroupedByMap = Maps.newIdentityHashMap();
    transaction.updates().forEach(update -> {
        AsyncConsistentMap<K, V> map = getMap(update.key());
        updatesGroupedByMap.computeIfAbsent(map, k -> Lists.newLinkedList()).add(update);
    });
    Map<AsyncConsistentMap<K, V>, MapTransaction<K, V>> transactionsByMap =
            Maps.transformValues(updatesGroupedByMap,
                                 list -> new MapTransaction<>(transaction.transactionId(), list));

    return Tools.allOf(transactionsByMap.entrySet()
                     .stream()
                     .map(e -> e.getKey().prepare(e.getValue()))
                     .collect(Collectors.toList()))
                .thenApply(list -> list.stream().reduce(Boolean::logicalAnd).orElse(true));
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
public CompletableFuture<Boolean> prepareAndCommit(MapTransaction<K, V> transaction) {
    Map<AsyncConsistentMap<K, V>, List<MapUpdate<K, V>>> updatesGroupedByMap = Maps.newIdentityHashMap();
    transaction.updates().forEach(update -> {
        AsyncConsistentMap<K, V> map = getMap(update.key());
        updatesGroupedByMap.computeIfAbsent(map, k -> Lists.newLinkedList()).add(update);
    });
    Map<AsyncConsistentMap<K, V>, MapTransaction<K, V>> transactionsByMap =
            Maps.transformValues(updatesGroupedByMap,
                                 list -> new MapTransaction<>(transaction.transactionId(), list));

    return Tools.allOf(transactionsByMap.entrySet()
                                        .stream()
                                        .map(e -> e.getKey().prepareAndCommit(e.getValue()))
                                        .collect(Collectors.toList()))
                .thenApply(list -> list.stream().reduce(Boolean::logicalAnd).orElse(true));
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
public FlowEntry getFlowEntry(FlowRule rule) {
    NodeId master = mastershipService.getMasterFor(rule.deviceId());

    if (master == null) {
        log.debug("Failed to getFlowEntry: No master for {}", rule.deviceId());
        return null;
    }

    if (Objects.equals(local, master)) {
        return flowTable.getFlowEntry(rule);
    }

    log.trace("Forwarding getFlowEntry to {}, which is the primary (master) for device {}",
              master, rule.deviceId());

    return Tools.futureGetOrElse(clusterCommunicator.sendAndReceive(rule,
                                FlowStoreMessageSubjects.GET_FLOW_ENTRY,
                                SERIALIZER::encode,
                                SERIALIZER::decode,
                                master),
                           FLOW_RULE_STORE_TIMEOUT_MILLIS,
                           TimeUnit.MILLISECONDS,
                           null);
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
public Iterable<FlowEntry> getFlowEntries(DeviceId deviceId) {
    NodeId master = mastershipService.getMasterFor(deviceId);

    if (master == null) {
        log.debug("Failed to getFlowEntries: No master for {}", deviceId);
        return Collections.emptyList();
    }

    if (Objects.equals(local, master)) {
        return flowTable.getFlowEntries(deviceId);
    }

    log.trace("Forwarding getFlowEntries to {}, which is the primary (master) for device {}",
              master, deviceId);

    return Tools.futureGetOrElse(clusterCommunicator.sendAndReceive(deviceId,
                                FlowStoreMessageSubjects.GET_DEVICE_FLOW_ENTRIES,
                                SERIALIZER::encode,
                                SERIALIZER::decode,
                                master),
                           FLOW_RULE_STORE_TIMEOUT_MILLIS,
                           TimeUnit.MILLISECONDS,
                           Collections.emptyList());
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
public <M> CompletableFuture<Void> unicast(M message,
                                           MessageSubject subject,
                                           Function<M, byte[]> encoder,
                                           NodeId toNodeId) {
    checkPermission(CLUSTER_WRITE);
    try {
        byte[] payload = new ClusterMessage(
                localNodeId,
                subject,
                timeFunction(encoder, subjectMeteringAgent, SERIALIZING).apply(message)
                ).getBytes();
        return doUnicast(subject, payload, toNodeId);
    } catch (Exception e) {
        return Tools.exceptionalFuture(e);
    }
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
public <M, R> CompletableFuture<R> sendAndReceive(M message,
                                                  MessageSubject subject,
                                                  Function<M, byte[]> encoder,
                                                  Function<byte[], R> decoder,
                                                  NodeId toNodeId) {
    checkPermission(CLUSTER_WRITE);
    try {
        ClusterMessage envelope = new ClusterMessage(
                clusterService.getLocalNode().id(),
                subject,
                timeFunction(encoder, subjectMeteringAgent, SERIALIZING).
                        apply(message));
        return sendAndReceive(subject, envelope.getBytes(), toNodeId).
                thenApply(bytes -> timeFunction(decoder, subjectMeteringAgent, DESERIALIZING).apply(bytes));
    } catch (Exception e) {
        return Tools.exceptionalFuture(e);
    }
}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
public Set<FlowEntry> getCurrentStatistic(ConnectPoint connectPoint) {
    final DeviceId deviceId = connectPoint.deviceId();
    NodeId master = mastershipService.getMasterFor(deviceId);
    if (master == null) {
        log.warn("No master for {}", deviceId);
        return Collections.emptySet();
    }
    if (master.equals(clusterService.getLocalNode().id())) {
        return getCurrentStatisticInternal(connectPoint);
    } else {
        return Tools.futureGetOrElse(clusterCommunicator.sendAndReceive(
                                    connectPoint,
                                    GET_CURRENT,
                                    SERIALIZER::encode,
                                    SERIALIZER::decode,
                                    master),
                               STATISTIC_STORE_TIMEOUT_MILLIS,
                               TimeUnit.MILLISECONDS,
                               Collections.emptySet());
    }

}
 

import org.onlab.util.Tools; //导入依赖的package包/类
@Override
public Set<FlowEntry> getPreviousStatistic(ConnectPoint connectPoint) {
    final DeviceId deviceId = connectPoint.deviceId();
    NodeId master = mastershipService.getMasterFor(deviceId);
    if (master == null) {
        log.warn("No master for {}", deviceId);
        return Collections.emptySet();
    }
    if (master.equals(clusterService.getLocalNode().id())) {
        return getPreviousStatisticInternal(connectPoint);
    } else {
        return Tools.futureGetOrElse(clusterCommunicator.sendAndReceive(
                                    connectPoint,
                                    GET_PREVIOUS,
                                    SERIALIZER::encode,
                                    SERIALIZER::decode,
                                    master),
                               STATISTIC_STORE_TIMEOUT_MILLIS,
                               TimeUnit.MILLISECONDS,
                               Collections.emptySet());
    }
}