import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Activate
public void activate() {
    tpool = Executors.newFixedThreadPool(4, groupedThreads("onos/flobj-notifier", "%d", log));
    eventQ = new LinkedBlockingQueue<ObjectiveEvent>();
    tpool.execute(new FlowObjectiveNotifier());
    nextGroups = storageService.<Integer, byte[]>consistentMapBuilder()
            .withName("flowobjective-groups")
            .withSerializer(Serializer.using(
                    new KryoNamespace.Builder()
                            .register(byte[].class)
                            .register(Versioned.class)
                            .build("DistributedFlowObjectiveStore")))
            .build();
    nextGroups.addListener(mapListener);
    nextIds = storageService.getAtomicCounter("next-objective-counter");
    log.info("Started");
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
<T> Stream<ContinuousResource> getAllocatedResources(DiscreteResourceId parent, Class<T> cls) {
    Set<ContinuousResource> children = getChildResources(parent);
    if (children.isEmpty()) {
        return Stream.of();
    }

    return children.stream()
            .filter(x -> x.id().equals(parent.child(cls)))
            // we don't use cascading simple predicates like follows to reduce accesses to consistent map
            // .filter(x -> continuousConsumers.containsKey(x.id()))
            // .filter(x -> continuousConsumers.get(x.id()) != null)
            // .filter(x -> !continuousConsumers.get(x.id()).value().allocations().isEmpty());
            .filter(resource -> {
                Versioned<ContinuousResourceAllocation> allocation = consumers.get(resource.id());
                if (allocation == null) {
                    return false;
                }
                return !allocation.value().allocations().isEmpty();
            });
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Activate
public void activate() {
    String localIp = getSiteLocalAddress();
    ControllerNode localNode =
            new DefaultControllerNode(new NodeId(localIp), IpAddress.valueOf(localIp), DEFAULT_ONOS_PORT);
    // partition 1
    Partition partition = new DefaultPartition(PartitionId.from(1), ImmutableSet.of(localNode.id()));
    ClusterMetadata metadata = new ClusterMetadata(PROVIDER_ID,
                                    "default",
                                    ImmutableSet.of(localNode),
                                    ImmutableSet.of(partition));
    long version = System.currentTimeMillis();
    cachedMetadata.set(new Versioned<>(metadata, version));
    providerRegistry.register(this);
    log.info("Started");
}
 

import org.onosproject.store.service.Versioned; //导入依赖的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.onosproject.store.service.Versioned; //导入依赖的package包/类
/**
 * Constructor to configure cache size.
 *
 * @param backingMap a distributed, strongly consistent map for backing
 * @param cacheSize the maximum size of the cache
 */
public CachingAsyncConsistentMap(AsyncConsistentMap<K, V> backingMap, int cacheSize) {
    super(backingMap);
    cache = CacheBuilder.newBuilder()
                        .maximumSize(cacheSize)
                        .build(CacheLoader.from(CachingAsyncConsistentMap.super::get));
    cacheUpdater = event -> {
        Versioned<V> newValue = event.newValue();
        if (newValue == null) {
            cache.invalidate(event.key());
        } else {
            cache.put(event.key(), CompletableFuture.completedFuture(newValue));
        }
    };
    statusListener = status -> {
        log.debug("{} status changed to {}", this.name(), status);
        // If the status of the underlying map is SUSPENDED or INACTIVE
        // we can no longer guarantee that the cache will be in sync.
        if (status == SUSPENDED || status == INACTIVE) {
            cache.invalidateAll();
        }
    };
    super.addListener(cacheUpdater);
    super.addStatusChangeListener(statusListener);
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Override
public void setDefaultContext(DeviceId deviceId) {
    Versioned<Bmv2DeviceContext> previous = contexts.put(deviceId, defaultContext);
    if (mastershipService.getMasterFor(deviceId) == null) {
        // Checking for who is the master here is ugly but necessary, as this method is called by Bmv2DeviceProvider
        // prior to master election. A solution could be to use a separate leadership contest instead of the
        // mastership service.
        triggerConfigCheck(deviceId, defaultContext);
    }
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Override
public void setPermissions(ApplicationId appId, Set<Permission> permissions) {
    AtomicBoolean permissionsChanged = new AtomicBoolean(false);
    Versioned<InternalApplicationHolder> appHolder = apps.computeIf(appId,
        v -> v != null && !Sets.symmetricDifference(v.permissions(), permissions).isEmpty(),
        (k, v) -> {
            permissionsChanged.set(true);
            return new InternalApplicationHolder(v.app(), v.state(), ImmutableSet.copyOf(permissions));
        });
    if (permissionsChanged.get()) {
        delegate.notify(new ApplicationEvent(APP_PERMISSIONS_CHANGED, appHolder.value().app()));
    }
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
Set<DiscreteResource> getChildResources(DiscreteResourceId parent) {
    Versioned<DiscreteResources> children = childMap.get(parent);

    if (children == null) {
        return ImmutableSet.of();
    }

    return children.value().values();
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Override
public long getFreeNumOfDevicePool(DeviceId deviceId) {
    Versioned<LabelResourcePool> pool = resourcePool.get(deviceId);
    if (pool == null) {
        return 0;
    }
    return pool.value().endLabel().labelId()
            - pool.value().currentUsedMaxLabelId().labelId()
            + pool.value().releaseLabelId().size();
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Override
public List<RuleId> getAllowingRuleByDenyingRule(RuleId denyingRuleId) {
    Versioned<List<RuleId>> allowRuleIdSet = denyRuleToAllowRule.get(denyingRuleId);
    if (allowRuleIdSet != null) {
        return allowRuleIdSet.value();
    } else {
        return null;
    }
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
List<ResourceAllocation> getResourceAllocations(ContinuousResourceId resource) {
    Versioned<ContinuousResourceAllocation> allocations = consumers.get(resource);
    if (allocations == null) {
        return ImmutableList.of();
    }

    return allocations.value().allocations().stream()
            .filter(x -> x.resource().id().equals(resource))
            .collect(GuavaCollectors.toImmutableList());
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Override
public CompletableFuture<V> getAndSet(V value) {
    final MeteringAgent.Context newTimer = monitor.startTimer(GET_AND_SET);
    if (value == null) {
        return backingMap.remove(name)
                         .thenApply(Versioned::valueOrNull)
                         .thenApply(v -> v == null ? null : serializer.<V>decode(v))
                         .whenComplete((r, e) -> newTimer.stop(e));
    }
    return backingMap.put(name, serializer.encode(value))
                     .thenApply(Versioned::valueOrNull)
                     .thenApply(v -> v == null ? null : serializer.<V>decode(v))
                     .whenComplete((r, e) -> newTimer.stop(e));
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Override
public Map<HostId, IpAssignment> listAllMapping() {
    Map<HostId, IpAssignment> validMapping = new HashMap<>();
    for (Map.Entry<HostId, Versioned<IpAssignment>> entry: allocationMap.entrySet()) {
        validMapping.put(entry.getKey(), entry.getValue().value());
    }
    return validMapping;
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Override
public V get(K key) {
    checkState(txContext.isOpen(), TX_CLOSED_ERROR);
    checkNotNull(key, ERROR_NULL_KEY);
    if (deleteSet.contains(key)) {
        return null;
    }
    V latest = writeCache.get(key);
    if (latest != null) {
        return latest;
    } else {
        Versioned<V> v = readCache.computeIfAbsent(key, k -> backingConsitentMap.get(k));
        return v != null ? v.value() : null;
    }
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
private void deactivateRequiredApps(ApplicationId appId) {
    getApplication(appId).requiredApps()
                         .stream()
                         .map(this::getId)
                         .map(apps::get)
                         .map(Versioned::value)
                         .filter(a -> a.state() == ACTIVATED)
                         .forEach(a -> deactivate(a.app().id(), appId));
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Override
public CompletableFuture<Versioned<V>> computeIfAbsent(K key,
                                                       Function<? super K, ? extends V> mappingFunction) {
    final MeteringAgent.Context timer = monitor.startTimer(COMPUTE_IF_ABSENT);
    return super.computeIfAbsent(key, mappingFunction)
                .whenComplete((r, e) -> timer.stop(e));
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
private void deactivateDependentApps(ApplicationId appId) {
    apps.values()
        .stream()
        .map(Versioned::value)
        .filter(a -> a.state() == ACTIVATED)
        .filter(a -> a.app().requiredApps().contains(appId.name()))
        .forEach(a -> deactivate(a.app().id()));
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
private void activate(ApplicationId appId, boolean updateTime) {
    Versioned<InternalApplicationHolder> vAppHolder = apps.get(appId);
    if (vAppHolder != null) {
        if (updateTime) {
            updateTime(appId.name());
        }
        activateRequiredApps(vAppHolder.value().app());

        apps.computeIf(appId, v -> v != null && v.state() != ACTIVATED,
                (k, v) -> new InternalApplicationHolder(
                        v.app(), ACTIVATED, v.permissions()));

    }
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Override
public CompletableFuture<Set<Entry<K1, Versioned<V1>>>> entrySet() {
    return backingMap.entrySet()
                     .thenApply(s -> s.stream()
                                      .map(e -> Maps.immutableEntry(keyDecoder.apply(e.getKey()),
                                                                    versionedValueTransform.apply(e.getValue())))
                                      .collect(Collectors.toSet()));
}
 

import org.onosproject.store.service.Versioned; //导入依赖的package包/类
@Override
public CompletableFuture<Versioned<V>> computeIf(K key,
        Predicate<? super V> condition,
        BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
    return super.computeIf(key, condition, remappingFunction)
            .whenComplete((r, e) -> cache.invalidate(key));
}