@Test public final void testUpdatePorts() { putDevice(DID1, SW1); List<PortDescription> pds = Arrays.<PortDescription>asList( new DefaultPortDescription(P1, true), new DefaultPortDescription(P2, true) ); Capture<InternalDeviceEvent> message = new Capture<>(); Capture<MessageSubject> subject = new Capture<>(); Capture<Function<InternalDeviceEvent, byte[]>> encoder = new Capture<>(); resetCommunicatorExpectingSingleBroadcast(message, subject, encoder); List<DeviceEvent> events = deviceStore.updatePorts(PID, DID1, pds); verify(clusterCommunicator); assertTrue(message.hasCaptured()); Set<PortNumber> expectedPorts = Sets.newHashSet(P1, P2); for (DeviceEvent event : events) { assertEquals(PORT_ADDED, event.type()); assertDevice(DID1, SW1, event.subject()); assertTrue("PortNumber is one of expected", expectedPorts.remove(event.port().number())); assertTrue("Port is enabled", event.port().isEnabled()); } assertTrue("Event for all expectedport appeared", expectedPorts.isEmpty()); List<PortDescription> pds2 = Arrays.<PortDescription>asList( new DefaultPortDescription(P1, false), new DefaultPortDescription(P2, true), new DefaultPortDescription(P3, true) ); resetCommunicatorExpectingSingleBroadcast(message, subject, encoder); events = deviceStore.updatePorts(PID, DID1, pds2); verify(clusterCommunicator); assertTrue(message.hasCaptured()); assertFalse("event should be triggered", events.isEmpty()); for (DeviceEvent event : events) { PortNumber num = event.port().number(); if (P1.equals(num)) { assertEquals(PORT_UPDATED, event.type()); assertDevice(DID1, SW1, event.subject()); assertFalse("Port is disabled", event.port().isEnabled()); } else if (P2.equals(num)) { fail("P2 event not expected."); } else if (P3.equals(num)) { assertEquals(PORT_ADDED, event.type()); assertDevice(DID1, SW1, event.subject()); assertTrue("Port is enabled", event.port().isEnabled()); } else { fail("Unknown port number encountered: " + num); } } List<PortDescription> pds3 = Arrays.<PortDescription>asList( new DefaultPortDescription(P1, false), new DefaultPortDescription(P2, true) ); resetCommunicatorExpectingSingleBroadcast(message, subject, encoder); events = deviceStore.updatePorts(PID, DID1, pds3); verify(clusterCommunicator); assertTrue(message.hasCaptured()); assertFalse("event should be triggered", events.isEmpty()); for (DeviceEvent event : events) { PortNumber num = event.port().number(); if (P1.equals(num)) { fail("P1 event not expected."); } else if (P2.equals(num)) { fail("P2 event not expected."); } else if (P3.equals(num)) { assertEquals(PORT_REMOVED, event.type()); assertDevice(DID1, SW1, event.subject()); assertTrue("Port was enabled", event.port().isEnabled()); } else { fail("Unknown port number encountered: " + num); } } }
@Override public synchronized List<DeviceEvent> updatePorts(ProviderId providerId, DeviceId deviceId, List<PortDescription> portDescriptions) { NodeId localNode = clusterService.getLocalNode().id(); NodeId deviceNode = mastershipService.getMasterFor(deviceId); List<DeviceEvent> deviceEvents = null; if (localNode.equals(deviceNode)) { final Timestamp newTimestamp; try { newTimestamp = deviceClockService.getTimestamp(deviceId); } catch (IllegalStateException e) { log.info("Timestamp was not available for device {}", deviceId); log.debug(" discarding {}", portDescriptions); return Collections.emptyList(); } log.debug("timestamp for {} {}", deviceId, newTimestamp); final Timestamped<List<PortDescription>> timestampedInput = new Timestamped<>(portDescriptions, newTimestamp); final Timestamped<List<PortDescription>> merged; final Map<ProviderId, DeviceDescriptions> device = getOrCreateDeviceDescriptionsMap(deviceId); synchronized (device) { deviceEvents = updatePortsInternal(providerId, deviceId, timestampedInput); final DeviceDescriptions descs = device.get(providerId); List<PortDescription> mergedList = FluentIterable.from(portDescriptions) .transform(input -> descs.getPortDesc(input.portNumber()).value() ).toList(); merged = new Timestamped<>(mergedList, newTimestamp); } if (!deviceEvents.isEmpty()) { log.debug("Notifying peers of a ports update topology event for providerId: {} and deviceId: {}", providerId, deviceId); notifyPeers(new InternalPortEvent(providerId, deviceId, merged)); } } else { if (!providerId.scheme().equals("cfg")) { return Collections.emptyList(); } if (deviceNode == null) { return Collections.emptyList(); } PortInjectedEvent portInjectedEvent = new PortInjectedEvent(providerId, deviceId, portDescriptions); clusterCommunicator.unicast(portInjectedEvent, PORT_INJECTED, SERIALIZER::encode, deviceNode); } return deviceEvents == null ? Collections.emptyList() : deviceEvents; }
GossipDeviceStore extends AbstractStore<DeviceEvent, DeviceStoreDelegate> implements DeviceStore { @Override public synchronized List<DeviceEvent> updatePorts(ProviderId providerId, DeviceId deviceId, List<PortDescription> portDescriptions) { NodeId localNode = clusterService.getLocalNode().id(); NodeId deviceNode = mastershipService.getMasterFor(deviceId); List<DeviceEvent> deviceEvents = null; if (localNode.equals(deviceNode)) { final Timestamp newTimestamp; try { newTimestamp = deviceClockService.getTimestamp(deviceId); } catch (IllegalStateException e) { log.info("Timestamp was not available for device {}", deviceId); log.debug(" discarding {}", portDescriptions); return Collections.emptyList(); } log.debug("timestamp for {} {}", deviceId, newTimestamp); final Timestamped<List<PortDescription>> timestampedInput = new Timestamped<>(portDescriptions, newTimestamp); final Timestamped<List<PortDescription>> merged; final Map<ProviderId, DeviceDescriptions> device = getOrCreateDeviceDescriptionsMap(deviceId); synchronized (device) { deviceEvents = updatePortsInternal(providerId, deviceId, timestampedInput); final DeviceDescriptions descs = device.get(providerId); List<PortDescription> mergedList = FluentIterable.from(portDescriptions) .transform(input -> descs.getPortDesc(input.portNumber()).value() ).toList(); merged = new Timestamped<>(mergedList, newTimestamp); } if (!deviceEvents.isEmpty()) { log.debug("Notifying peers of a ports update topology event for providerId: {} and deviceId: {}", providerId, deviceId); notifyPeers(new InternalPortEvent(providerId, deviceId, merged)); } } else { if (!providerId.scheme().equals("cfg")) { return Collections.emptyList(); } if (deviceNode == null) { return Collections.emptyList(); } PortInjectedEvent portInjectedEvent = new PortInjectedEvent(providerId, deviceId, portDescriptions); clusterCommunicator.unicast(portInjectedEvent, PORT_INJECTED, SERIALIZER::encode, deviceNode); } return deviceEvents == null ? Collections.emptyList() : deviceEvents; } }
GossipDeviceStore extends AbstractStore<DeviceEvent, DeviceStoreDelegate> implements DeviceStore { @Override public synchronized List<DeviceEvent> updatePorts(ProviderId providerId, DeviceId deviceId, List<PortDescription> portDescriptions) { NodeId localNode = clusterService.getLocalNode().id(); NodeId deviceNode = mastershipService.getMasterFor(deviceId); List<DeviceEvent> deviceEvents = null; if (localNode.equals(deviceNode)) { final Timestamp newTimestamp; try { newTimestamp = deviceClockService.getTimestamp(deviceId); } catch (IllegalStateException e) { log.info("Timestamp was not available for device {}", deviceId); log.debug(" discarding {}", portDescriptions); return Collections.emptyList(); } log.debug("timestamp for {} {}", deviceId, newTimestamp); final Timestamped<List<PortDescription>> timestampedInput = new Timestamped<>(portDescriptions, newTimestamp); final Timestamped<List<PortDescription>> merged; final Map<ProviderId, DeviceDescriptions> device = getOrCreateDeviceDescriptionsMap(deviceId); synchronized (device) { deviceEvents = updatePortsInternal(providerId, deviceId, timestampedInput); final DeviceDescriptions descs = device.get(providerId); List<PortDescription> mergedList = FluentIterable.from(portDescriptions) .transform(input -> descs.getPortDesc(input.portNumber()).value() ).toList(); merged = new Timestamped<>(mergedList, newTimestamp); } if (!deviceEvents.isEmpty()) { log.debug("Notifying peers of a ports update topology event for providerId: {} and deviceId: {}", providerId, deviceId); notifyPeers(new InternalPortEvent(providerId, deviceId, merged)); } } else { if (!providerId.scheme().equals("cfg")) { return Collections.emptyList(); } if (deviceNode == null) { return Collections.emptyList(); } PortInjectedEvent portInjectedEvent = new PortInjectedEvent(providerId, deviceId, portDescriptions); clusterCommunicator.unicast(portInjectedEvent, PORT_INJECTED, SERIALIZER::encode, deviceNode); } return deviceEvents == null ? Collections.emptyList() : deviceEvents; }  }
GossipDeviceStore extends AbstractStore<DeviceEvent, DeviceStoreDelegate> implements DeviceStore { @Override public synchronized List<DeviceEvent> updatePorts(ProviderId providerId, DeviceId deviceId, List<PortDescription> portDescriptions) { NodeId localNode = clusterService.getLocalNode().id(); NodeId deviceNode = mastershipService.getMasterFor(deviceId); List<DeviceEvent> deviceEvents = null; if (localNode.equals(deviceNode)) { final Timestamp newTimestamp; try { newTimestamp = deviceClockService.getTimestamp(deviceId); } catch (IllegalStateException e) { log.info("Timestamp was not available for device {}", deviceId); log.debug(" discarding {}", portDescriptions); return Collections.emptyList(); } log.debug("timestamp for {} {}", deviceId, newTimestamp); final Timestamped<List<PortDescription>> timestampedInput = new Timestamped<>(portDescriptions, newTimestamp); final Timestamped<List<PortDescription>> merged; final Map<ProviderId, DeviceDescriptions> device = getOrCreateDeviceDescriptionsMap(deviceId); synchronized (device) { deviceEvents = updatePortsInternal(providerId, deviceId, timestampedInput); final DeviceDescriptions descs = device.get(providerId); List<PortDescription> mergedList = FluentIterable.from(portDescriptions) .transform(input -> descs.getPortDesc(input.portNumber()).value() ).toList(); merged = new Timestamped<>(mergedList, newTimestamp); } if (!deviceEvents.isEmpty()) { log.debug("Notifying peers of a ports update topology event for providerId: {} and deviceId: {}", providerId, deviceId); notifyPeers(new InternalPortEvent(providerId, deviceId, merged)); } } else { if (!providerId.scheme().equals("cfg")) { return Collections.emptyList(); } if (deviceNode == null) { return Collections.emptyList(); } PortInjectedEvent portInjectedEvent = new PortInjectedEvent(providerId, deviceId, portDescriptions); clusterCommunicator.unicast(portInjectedEvent, PORT_INJECTED, SERIALIZER::encode, deviceNode); } return deviceEvents == null ? Collections.emptyList() : deviceEvents; }  @Activate void activate(); @Deactivate void deactivate(); @Override int getDeviceCount(); @Override Iterable<Device> getDevices(); @Override Iterable<Device> getAvailableDevices(); @Override Device getDevice(DeviceId deviceId); @Override synchronized DeviceEvent createOrUpdateDevice(ProviderId providerId,
                                                         DeviceId deviceId,
                                                         DeviceDescription deviceDescription); @Override DeviceEvent markOffline(DeviceId deviceId); @Override synchronized List<DeviceEvent> updatePorts(ProviderId providerId,
                                                      DeviceId deviceId,
                                                      List<PortDescription> portDescriptions); @Override synchronized DeviceEvent updatePortStatus(ProviderId providerId,
                                                     DeviceId deviceId,
                                                     PortDescription portDescription); @Override List<Port> getPorts(DeviceId deviceId); @Override DeviceEvent updatePortStatistics(ProviderId providerId, DeviceId deviceId,
                                            Collection<PortStatistics> newStatsCollection); PortStatistics calcDeltaStats(DeviceId deviceId, PortStatistics prvStats, PortStatistics newStats); @Override List<PortStatistics> getPortStatistics(DeviceId deviceId); @Override List<PortStatistics> getPortDeltaStatistics(DeviceId deviceId); @Override Port getPort(DeviceId deviceId, PortNumber portNumber); @Override boolean isAvailable(DeviceId deviceId); @Override synchronized DeviceEvent removeDevice(DeviceId deviceId); }
GossipDeviceStore extends AbstractStore<DeviceEvent, DeviceStoreDelegate> implements DeviceStore { @Override public synchronized List<DeviceEvent> updatePorts(ProviderId providerId, DeviceId deviceId, List<PortDescription> portDescriptions) { NodeId localNode = clusterService.getLocalNode().id(); NodeId deviceNode = mastershipService.getMasterFor(deviceId); List<DeviceEvent> deviceEvents = null; if (localNode.equals(deviceNode)) { final Timestamp newTimestamp; try { newTimestamp = deviceClockService.getTimestamp(deviceId); } catch (IllegalStateException e) { log.info("Timestamp was not available for device {}", deviceId); log.debug(" discarding {}", portDescriptions); return Collections.emptyList(); } log.debug("timestamp for {} {}", deviceId, newTimestamp); final Timestamped<List<PortDescription>> timestampedInput = new Timestamped<>(portDescriptions, newTimestamp); final Timestamped<List<PortDescription>> merged; final Map<ProviderId, DeviceDescriptions> device = getOrCreateDeviceDescriptionsMap(deviceId); synchronized (device) { deviceEvents = updatePortsInternal(providerId, deviceId, timestampedInput); final DeviceDescriptions descs = device.get(providerId); List<PortDescription> mergedList = FluentIterable.from(portDescriptions) .transform(input -> descs.getPortDesc(input.portNumber()).value() ).toList(); merged = new Timestamped<>(mergedList, newTimestamp); } if (!deviceEvents.isEmpty()) { log.debug("Notifying peers of a ports update topology event for providerId: {} and deviceId: {}", providerId, deviceId); notifyPeers(new InternalPortEvent(providerId, deviceId, merged)); } } else { if (!providerId.scheme().equals("cfg")) { return Collections.emptyList(); } if (deviceNode == null) { return Collections.emptyList(); } PortInjectedEvent portInjectedEvent = new PortInjectedEvent(providerId, deviceId, portDescriptions); clusterCommunicator.unicast(portInjectedEvent, PORT_INJECTED, SERIALIZER::encode, deviceNode); } return deviceEvents == null ? Collections.emptyList() : deviceEvents; }  @Activate void activate(); @Deactivate void deactivate(); @Override int getDeviceCount(); @Override Iterable<Device> getDevices(); @Override Iterable<Device> getAvailableDevices(); @Override Device getDevice(DeviceId deviceId); @Override synchronized DeviceEvent createOrUpdateDevice(ProviderId providerId,
                                                         DeviceId deviceId,
                                                         DeviceDescription deviceDescription); @Override DeviceEvent markOffline(DeviceId deviceId); @Override synchronized List<DeviceEvent> updatePorts(ProviderId providerId,
                                                      DeviceId deviceId,
                                                      List<PortDescription> portDescriptions); @Override synchronized DeviceEvent updatePortStatus(ProviderId providerId,
                                                     DeviceId deviceId,
                                                     PortDescription portDescription); @Override List<Port> getPorts(DeviceId deviceId); @Override DeviceEvent updatePortStatistics(ProviderId providerId, DeviceId deviceId,
                                            Collection<PortStatistics> newStatsCollection); PortStatistics calcDeltaStats(DeviceId deviceId, PortStatistics prvStats, PortStatistics newStats); @Override List<PortStatistics> getPortStatistics(DeviceId deviceId); @Override List<PortStatistics> getPortDeltaStatistics(DeviceId deviceId); @Override Port getPort(DeviceId deviceId, PortNumber portNumber); @Override boolean isAvailable(DeviceId deviceId); @Override synchronized DeviceEvent removeDevice(DeviceId deviceId);  }