import org.onosproject.cluster.NodeId; //导入依赖的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.onosproject.cluster.NodeId; //导入依赖的package包/类
@Override
public RoleInfo decode(ObjectNode json, CodecContext context) {
    if (json == null || !json.isObject()) {
        return null;
    }

    // parse node identifier of master
    NodeId nodeId = json.get(MASTER) == null ?
            null : NodeId.nodeId(json.get(MASTER).asText());

    // parse node identifier of backups
    List<NodeId> backups = new ArrayList<>();

    ArrayNode backupsJson = (ArrayNode) nullIsIllegal(json.get(BACKUPS),
            BACKUPS + MISSING_MEMBER_MESSAGE);

    IntStream.range(0, backupsJson.size()).forEach(i -> {
        JsonNode backupJson = nullIsIllegal(backupsJson.get(i),
                "Backup node id cannot be null");
        backups.add(NodeId.nodeId(backupJson.asText()));
    });

    return new RoleInfo(nodeId, backups);
}
 

import org.onosproject.cluster.NodeId; //导入依赖的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());
    }
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
@Override
protected void execute() {
    RegionService regionService = get(RegionService.class);
    RegionAdminService regionAdminService = get(RegionAdminService.class);
    RegionId regionId = RegionId.regionId(id);

    if (regionService.getRegion(regionId) == null) {
        print("The region with id %s does not exist.", regionId);
        return;
    }

    List<Set<NodeId>> masters = Lists.newArrayList();
    Set<NodeId> nodeIds = Sets.newHashSet();
    for (String masterArg : masterArgs) {
        if (masterArg.equals("/")) {
            masters.add(nodeIds);
            nodeIds = Sets.newHashSet();
        } else {
            nodeIds.add(NodeId.nodeId(masterArg));
        }
    }
    masters.add(nodeIds);

    regionAdminService.updateRegion(regionId, name, REGION_TYPE_MAP.get(type), masters);
    print("Region with id %s is successfully updated.", regionId);
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
private void checkDeviceMasters(Set<DeviceId> deviceIds, Set<NodeId> expectedMasters,
                                 Consumer<DeviceId> checkRole) {
    // each device's master must be contained in the list of expectedMasters
    deviceIds.stream().forEach(deviceId -> {
        assertTrue("wrong master:", expectedMasters.contains(mgr.getMasterFor(deviceId)));
        if (checkRole != null) {
            checkRole.accept(deviceId);
        }
    });
    // each node in expectedMasters must have approximately the same number of devices
    if (expectedMasters.size() > 1) {
        int minValue = Integer.MAX_VALUE;
        int maxDevices = -1;
        for (NodeId nodeId: expectedMasters) {
            int numDevicesManagedByNode = mgr.getDevicesOf(nodeId).size();
            if (numDevicesManagedByNode < minValue) {
                minValue = numDevicesManagedByNode;
            }
            if (numDevicesManagedByNode > maxDevices) {
                maxDevices = numDevicesManagedByNode;
            }
            assertTrue("not balanced:", maxDevices - minValue <= 1);
        }
    }
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
private void sendAdvertisementToPeer(NodeId peer) {
    long adCreationTime = System.currentTimeMillis();
    AntiEntropyAdvertisement<K> ad = createAdvertisement();
    clusterCommunicator.sendAndReceive(ad,
            antiEntropyAdvertisementSubject,
            serializer::encode,
            serializer::decode,
            peer)
            .whenComplete((result, error) -> {
                if (error != null) {
                    log.debug("Failed to send anti-entropy advertisement to {}", peer, error);
                } else if (result == AntiEntropyResponse.PROCESSED) {
                    antiEntropyTimes.put(peer, adCreationTime);
                }
            });
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
/**
 * Sets up the global values for all the tests.
 */
@Before
public void setUpTest() {
    final CodecManager codecService = new CodecManager();
    codecService.activate();
    codecService.registerCodec(ControlLoadSnapshot.class, new ControlLoadSnapshotCodec());
    ServiceDirectory testDirectory =
            new TestServiceDirectory()
                    .add(ControlPlaneMonitorService.class,
                            mockControlPlaneMonitorService)
                    .add(ClusterService.class, mockClusterService)
                    .add(CodecService.class, codecService);
    BaseResource.setServiceDirectory(testDirectory);

    nodeId = new NodeId("1");
    mockControlLoad = new MockControlLoad();
    ControllerNode mockControllerNode = new MockControllerNode(nodeId);

    expect(mockClusterService.getLocalNode()).andReturn(mockControllerNode).anyTimes();
    replay(mockClusterService);
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
@Override
public MastershipTerm decode(ObjectNode json, CodecContext context) {
    if (json == null || !json.isObject()) {
        return null;
    }

    // node identifier of master
    NodeId nodeId = NodeId.nodeId(nullIsIllegal(json.get(MASTER),
            MASTER + MISSING_MEMBER_MESSAGE).asText());

    // term number
    long termNumber = nullIsIllegal(json.get(TERM_NUMBER),
            TERM_NUMBER + MISSING_MEMBER_MESSAGE).asLong();

    return MastershipTerm.of(nodeId, termNumber);
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
public ElectionState promote(NodeId nodeId) {
    Registration registration = registrations.stream()
                                          .filter(r -> r.nodeId().equals(nodeId))
                                          .findFirst()
                                          .orElse(null);
    List<Registration> updatedRegistrations = Lists.newArrayList();
    updatedRegistrations.add(registration);
    registrations.stream()
                 .filter(r -> !r.nodeId().equals(nodeId))
                 .forEach(updatedRegistrations::add);
    return new ElectionState(updatedRegistrations,
                            leader,
                            term,
                            termStartTime);

}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
private void assertInternalDeviceEvent(NodeId sender,
        DeviceId deviceId,
        ProviderId providerId,
        DeviceDescription expectedDesc,
        Capture<InternalDeviceEvent> actualEvent,
        Capture<MessageSubject> actualSubject,
        Capture<Function<InternalDeviceEvent, byte[]>> actualEncoder) {
    assertTrue(actualEvent.hasCaptured());
    assertTrue(actualSubject.hasCaptured());
    assertTrue(actualEncoder.hasCaptured());

    assertEquals(GossipDeviceStoreMessageSubjects.DEVICE_UPDATE,
            actualSubject.getValue());
    assertEquals(deviceId, actualEvent.getValue().deviceId());
    assertEquals(providerId, actualEvent.getValue().providerId());
    assertDeviceDescriptionEquals(expectedDesc, actualEvent.getValue().deviceDescription().value());
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
public ElectionState evict(NodeId nodeId, Supplier<Long> termCounter) {
    Optional<Registration> registration =
            registrations.stream().filter(r -> r.nodeId.equals(nodeId)).findFirst();
    if (registration.isPresent()) {
        List<Registration> updatedRegistrations =
                registrations.stream()
                .filter(r -> !r.nodeId().equals(nodeId))
                .collect(Collectors.toList());
        if (leader.nodeId().equals(nodeId)) {
            if (updatedRegistrations.size() > 0) {
                return new ElectionState(updatedRegistrations,
                        updatedRegistrations.get(0),
                        termCounter.get(),
                        System.currentTimeMillis());
            } else {
                return new ElectionState(updatedRegistrations, null, term, termStartTime);
            }
        } else {
            return new ElectionState(updatedRegistrations, leader, term, termStartTime);
        }
    } else {
        return this;
    }
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
@Override
public Region createRegion(RegionId regionId, String name, Region.Type type,
                           List<Set<NodeId>> masterNodeIds) {
    checkNotNull(regionId, REGION_ID_NULL);
    checkNotNull(name, NAME_NULL);
    checkNotNull(name, REGION_TYPE_NULL);
    return store.createRegion(regionId, name, type, masterNodeIds == null ? of() : masterNodeIds);
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
public DeviceAntiEntropyAdvertisement(NodeId sender,
            Map<DeviceFragmentId, Timestamp> devices,
            Map<PortFragmentId, Timestamp> ports,
            Map<DeviceId, Timestamp> offline) {
    this.sender = checkNotNull(sender);
    this.deviceFingerPrints = checkNotNull(devices);
    this.portFingerPrints = checkNotNull(ports);
    this.offline = checkNotNull(offline);
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
@Override
public FlowRuleEvent addOrUpdateFlowRule(FlowEntry rule) {
    NodeId master = mastershipService.getMasterFor(rule.deviceId());
    if (Objects.equals(local, master)) {
        return addOrUpdateFlowRuleInternal(rule);
    }

    log.warn("Tried to update FlowRule {} state,"
                     + " while the Node was not the master.", rule);
    return null;
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
/**
 * Tests decoding of a json object.
 */
@Test
public void testRegionDecode() throws IOException {
    Region region = getRegion("Region.json");
    checkCommonData(region);

    assertThat(region.masters().size(), is(2));

    NodeId nodeId1 = NodeId.nodeId("1");
    NodeId nodeId2 = NodeId.nodeId("2");
    Set<NodeId> nodeIds1 = region.masters().get(0);
    Set<NodeId> nodeIds2 = region.masters().get(1);
    assertThat(nodeIds1.containsAll(ImmutableSet.of(nodeId1)), is(true));
    assertThat(nodeIds2.containsAll(ImmutableSet.of(nodeId1, nodeId2)), is(true));
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
@Override
public NodeId getMasterFor(DeviceId deviceId) {
    if (DEVS_TRUNK.contains(deviceId)) {
        return CNID_1;
    }
    if (DEVS_LEFT.contains(deviceId)) {
        return CNID_2;
    }
    if (DEVS_RIGHT.contains(deviceId)) {
        return CNID_3;
    }
    return null;
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
@Test
public void basics() {
    ImmutableList<Set<NodeId>> masters = ImmutableList
            .of(ImmutableSet.of(nodeId("n1"), nodeId("n2")),
                ImmutableSet.of(nodeId("n3"), nodeId("n4")));
    Region r = new DefaultRegion(ID1, "R1", METRO, masters);
    assertEquals("incorrect id", ID1, r.id());
    assertEquals("incorrect name", "R1", r.name());
    assertEquals("incorrect type", METRO, r.type());
    assertEquals("incorrect masters", masters, r.masters());
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
private Set<NodeId> jsonToSet(JsonNode nodes) {
    final Set<NodeId> nodeIds = Sets.newHashSet();
    nodes.forEach(node -> nodeIds.add(NodeId.nodeId(node.asText())));
    return nodeIds;
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
private synchronized void addSample(long time, NodeId nodeId,
                                    double overallRate, double currentRate) {
    Sample fullSample = createCurrentSampleIfNeeded(time);
    setSampleData(current, nodeId, currentRate);
    setSampleData(overall, nodeId, overallRate);
    pruneSamplesIfNeeded();

    if (fullSample != null && ui != null) {
        ui.reportSample(fullSample);
    }
}
 

import org.onosproject.cluster.NodeId; //导入依赖的package包/类
@Override
public boolean destroyDevicePool(DeviceId deviceId) {
    Device device = deviceService.getDevice(deviceId);
    if (device == null) {
        return false;
    }

    NodeId master = mastershipService.getMasterFor(deviceId);

    if (master == null) {
        log.warn("Failed to destroyDevicePool. No master for {}", deviceId);
        return false;
    }

    if (master.equals(clusterService.getLocalNode().id())) {
        return internalDestroy(deviceId);
    }

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

    return complete(clusterCommunicator
            .sendAndReceive(deviceId,
                            LabelResourceMessageSubjects.LABEL_POOL_DESTROYED,
                            SERIALIZER::encode, SERIALIZER::decode,
                            master));
}