@Test public void testMetadataHaUseCase() { AddressEndpointBuilder mac1 = getAddressEndpointBuilder().setKey(MAC_KEY_23_45); setParentEndpoints(mac1, IP_KEY_1_22, IP_KEY_9_99); AddressEndpointBuilder mac2 = getAddressEndpointBuilder().setKey(MAC_KEY_24_45); setParentEndpoints(mac2, IP_KEY_2_22, IP_KEY_9_99); AddressEndpointBuilder mac3 = getAddressEndpointBuilder().setKey(MAC_KEY_25_45); setParentEndpoints(mac3, IP_KEY_3_22, IP_KEY_9_99); AddressEndpointBuilder ip1 = getAddressEndpointBuilder().setKey(IP_KEY_1_22); setChildEndpoints(ip1, MAC_KEY_23_45); AddressEndpointBuilder ip2 = getAddressEndpointBuilder().setKey(IP_KEY_2_22); setChildEndpoints(ip2, MAC_KEY_24_45); AddressEndpointBuilder ip3 = getAddressEndpointBuilder().setKey(IP_KEY_3_22); setChildEndpoints(ip3, MAC_KEY_25_45); AddressEndpointBuilder ip9 = getAddressEndpointBuilder().setKey(IP_KEY_9_99); setChildEndpoints(ip9, MAC_KEY_23_45, MAC_KEY_24_45, MAC_KEY_25_45); submitEndpointsToDatastore(mac1.build(), mac2.build(), mac3.build(), ip1.build(), ip2.build(), ip3.build(), ip9.build()); assertEndpointsInDatastore(mac1.getKey(), mac2.getKey(), mac3.getKey(), ip1.getKey(), ip2.getKey(), ip3.getKey(), ip9.getKey()); locationProvider.createLocationForVppEndpoint(vppEndpointBuilder(MAC_KEY_23_45_VPP, NODE_1, INTF_NAME)); locationProvider.createLocationForVppEndpoint(vppEndpointBuilder(MAC_KEY_24_45_VPP, NODE_2, INTF_NAME)); locationProvider.createLocationForVppEndpoint(vppEndpointBuilder(MAC_KEY_25_45_VPP, NODE_3, INTF_NAME)); assertL2ChildHasTwoIpParentsAndIpParentHasMultipleChilds(4L); }
@Override public ListenableFuture<Void> createLocationForVppEndpoint(VppEndpoint vppEndpoint) { try { THREAD_LOCK.lock(); vppEndpoints.put(vppEndpoint.getKey(), vppEndpoint); Collection<AddressEndpoint> cachedAddrEp = pendingAddrEndpoints.row(vppEndpoint.getKey()).values(); if (cachedAddrEp == null || cachedAddrEp.isEmpty()) { return Futures.immediateFuture(null); } ReadWriteTransaction rwTx = syncedChain.newReadWriteTransaction(); List<AddressEndpoint> result = cachedAddrEp.stream().filter(ep -> canCreateLocation(ep)).collect(Collectors.toList()); result.forEach(ep -> syncEndpointLocation(rwTx, ep)); result.forEach(ep -> pendingAddrEndpoints.remove(vppEndpoint.getKey(), ep.getKey())); syncedChain.submitNow(rwTx); } catch (Exception e) { LOG.error("Failed to resolve location for vpp endpoint {}. {}", vppEndpoint.getKey(), e); } finally { THREAD_LOCK.unlock(); } return Futures.immediateFuture(null); }
VppEndpointLocationProvider implements ClusteredDataTreeChangeListener<AddressEndpoint>, VPPLocationProvider, AutoCloseable { @Override public ListenableFuture<Void> createLocationForVppEndpoint(VppEndpoint vppEndpoint) { try { THREAD_LOCK.lock(); vppEndpoints.put(vppEndpoint.getKey(), vppEndpoint); Collection<AddressEndpoint> cachedAddrEp = pendingAddrEndpoints.row(vppEndpoint.getKey()).values(); if (cachedAddrEp == null || cachedAddrEp.isEmpty()) { return Futures.immediateFuture(null); } ReadWriteTransaction rwTx = syncedChain.newReadWriteTransaction(); List<AddressEndpoint> result = cachedAddrEp.stream().filter(ep -> canCreateLocation(ep)).collect(Collectors.toList()); result.forEach(ep -> syncEndpointLocation(rwTx, ep)); result.forEach(ep -> pendingAddrEndpoints.remove(vppEndpoint.getKey(), ep.getKey())); syncedChain.submitNow(rwTx); } catch (Exception e) { LOG.error("Failed to resolve location for vpp endpoint {}. {}", vppEndpoint.getKey(), e); } finally { THREAD_LOCK.unlock(); } return Futures.immediateFuture(null); } }
VppEndpointLocationProvider implements ClusteredDataTreeChangeListener<AddressEndpoint>, VPPLocationProvider, AutoCloseable { @Override public ListenableFuture<Void> createLocationForVppEndpoint(VppEndpoint vppEndpoint) { try { THREAD_LOCK.lock(); vppEndpoints.put(vppEndpoint.getKey(), vppEndpoint); Collection<AddressEndpoint> cachedAddrEp = pendingAddrEndpoints.row(vppEndpoint.getKey()).values(); if (cachedAddrEp == null || cachedAddrEp.isEmpty()) { return Futures.immediateFuture(null); } ReadWriteTransaction rwTx = syncedChain.newReadWriteTransaction(); List<AddressEndpoint> result = cachedAddrEp.stream().filter(ep -> canCreateLocation(ep)).collect(Collectors.toList()); result.forEach(ep -> syncEndpointLocation(rwTx, ep)); result.forEach(ep -> pendingAddrEndpoints.remove(vppEndpoint.getKey(), ep.getKey())); syncedChain.submitNow(rwTx); } catch (Exception e) { LOG.error("Failed to resolve location for vpp endpoint {}. {}", vppEndpoint.getKey(), e); } finally { THREAD_LOCK.unlock(); } return Futures.immediateFuture(null); } VppEndpointLocationProvider(DataBroker dataProvider); }
VppEndpointLocationProvider implements ClusteredDataTreeChangeListener<AddressEndpoint>, VPPLocationProvider, AutoCloseable { @Override public ListenableFuture<Void> createLocationForVppEndpoint(VppEndpoint vppEndpoint) { try { THREAD_LOCK.lock(); vppEndpoints.put(vppEndpoint.getKey(), vppEndpoint); Collection<AddressEndpoint> cachedAddrEp = pendingAddrEndpoints.row(vppEndpoint.getKey()).values(); if (cachedAddrEp == null || cachedAddrEp.isEmpty()) { return Futures.immediateFuture(null); } ReadWriteTransaction rwTx = syncedChain.newReadWriteTransaction(); List<AddressEndpoint> result = cachedAddrEp.stream().filter(ep -> canCreateLocation(ep)).collect(Collectors.toList()); result.forEach(ep -> syncEndpointLocation(rwTx, ep)); result.forEach(ep -> pendingAddrEndpoints.remove(vppEndpoint.getKey(), ep.getKey())); syncedChain.submitNow(rwTx); } catch (Exception e) { LOG.error("Failed to resolve location for vpp endpoint {}. {}", vppEndpoint.getKey(), e); } finally { THREAD_LOCK.unlock(); } return Futures.immediateFuture(null); } VppEndpointLocationProvider(DataBroker dataProvider); @Override void onDataTreeChanged(Collection<DataTreeModification<AddressEndpoint>> change); @Override ListenableFuture<Void> createLocationForVppEndpoint(VppEndpoint vppEndpoint); @Override ListenableFuture<Void> deleteLocationForVppEndpoint(VppEndpoint vppEndpoint); @Override ListenableFuture<Void> replaceLocationForEndpoint(@Nonnull ExternalLocationCase location,
            @Nonnull AddressEndpointWithLocationKey addrEpWithLocKey); @Override void close(); }
VppEndpointLocationProvider implements ClusteredDataTreeChangeListener<AddressEndpoint>, VPPLocationProvider, AutoCloseable { @Override public ListenableFuture<Void> createLocationForVppEndpoint(VppEndpoint vppEndpoint) { try { THREAD_LOCK.lock(); vppEndpoints.put(vppEndpoint.getKey(), vppEndpoint); Collection<AddressEndpoint> cachedAddrEp = pendingAddrEndpoints.row(vppEndpoint.getKey()).values(); if (cachedAddrEp == null || cachedAddrEp.isEmpty()) { return Futures.immediateFuture(null); } ReadWriteTransaction rwTx = syncedChain.newReadWriteTransaction(); List<AddressEndpoint> result = cachedAddrEp.stream().filter(ep -> canCreateLocation(ep)).collect(Collectors.toList()); result.forEach(ep -> syncEndpointLocation(rwTx, ep)); result.forEach(ep -> pendingAddrEndpoints.remove(vppEndpoint.getKey(), ep.getKey())); syncedChain.submitNow(rwTx); } catch (Exception e) { LOG.error("Failed to resolve location for vpp endpoint {}. {}", vppEndpoint.getKey(), e); } finally { THREAD_LOCK.unlock(); } return Futures.immediateFuture(null); } VppEndpointLocationProvider(DataBroker dataProvider); @Override void onDataTreeChanged(Collection<DataTreeModification<AddressEndpoint>> change); @Override ListenableFuture<Void> createLocationForVppEndpoint(VppEndpoint vppEndpoint); @Override ListenableFuture<Void> deleteLocationForVppEndpoint(VppEndpoint vppEndpoint); @Override ListenableFuture<Void> replaceLocationForEndpoint(@Nonnull ExternalLocationCase location,
            @Nonnull AddressEndpointWithLocationKey addrEpWithLocKey); @Override void close(); static final ProviderName VPP_ENDPOINT_LOCATION_PROVIDER; static final long PROVIDER_PRIORITY; }