AddressObservationWriter { public void addAddress(MacAddress macAddress, IpAddress ipAddress, NodeConnectorRef nodeConnectorRef) { if (macAddress == null || ipAddress == null || nodeConnectorRef == null) { return; } NodeConnectorLock nodeConnectorLock = lockMap.computeIfAbsent(nodeConnectorRef, key -> new NodeConnectorLock()); synchronized (nodeConnectorLock) { ListenableFuture<Void> future = futureMap.get(nodeConnectorLock); if (future != null) { try { future.get(); } catch (InterruptedException | ExecutionException e) { LOG.error("Exception while waiting for previous transaction to finish", e); } } long now = new Date().getTime(); final AddressCapableNodeConnectorBuilder acncBuilder = new AddressCapableNodeConnectorBuilder(); final AddressesBuilder addressBuilder = new AddressesBuilder().setIp(ipAddress).setMac(macAddress) .setFirstSeen(now).setLastSeen(now); List<Addresses> addresses = null; ReadOnlyTransaction readTransaction = dataService.newReadOnlyTransaction(); NodeConnector nc = null; try { Optional<NodeConnector> dataObjectOptional = readTransaction.read(LogicalDatastoreType.OPERATIONAL, (InstanceIdentifier<NodeConnector>) nodeConnectorRef.getValue()).get(); if (dataObjectOptional.isPresent()) { nc = dataObjectOptional.get(); } } catch (InterruptedException | ExecutionException e) { LOG.error("Error reading node connector {}", nodeConnectorRef.getValue()); readTransaction.close(); throw new RuntimeException("Error reading from operational store, node connector : " + nodeConnectorRef, e); } readTransaction.close(); if (nc == null) { return; } AddressCapableNodeConnector acnc = nc .getAugmentation(AddressCapableNodeConnector.class); if (acnc != null && acnc.getAddresses() != null) { addresses = acnc.getAddresses(); for (int i = 0; i < addresses.size(); i++) { if (addresses.get(i).getIp().equals(ipAddress) && addresses.get(i).getMac().equals(macAddress)) { if (now - addresses.get(i).getLastSeen() > timestampUpdateInterval) { addressBuilder.setFirstSeen(addresses.get(i).getFirstSeen()) .setKey(addresses.get(i).getKey()); addresses.remove(i); break; } else { return; } } } } else { addresses = new ArrayList<>(); } if (addressBuilder.getKey() == null) { addressBuilder.setKey(new AddressesKey(BigInteger.valueOf(addressKey.getAndIncrement()))); } addresses.add(addressBuilder.build()); acncBuilder.setAddresses(addresses); InstanceIdentifier<AddressCapableNodeConnector> addressCapableNcInstanceId = ((InstanceIdentifier<NodeConnector>) nodeConnectorRef .getValue()).augmentation(AddressCapableNodeConnector.class); final WriteTransaction writeTransaction = dataService.newWriteOnlyTransaction(); writeTransaction.merge(LogicalDatastoreType.OPERATIONAL, addressCapableNcInstanceId, acncBuilder.build()); final ListenableFuture<Void> writeTxResultFuture = writeTransaction.submit(); Futures.addCallback(writeTxResultFuture, new FutureCallback<Void>() { @Override public void onSuccess(Void notUsed) { LOG.debug("AddressObservationWriter write successful for tx :{}", writeTransaction.getIdentifier()); } @Override public void onFailure(Throwable throwable) { LOG.error("AddressObservationWriter write transaction {} failed", writeTransaction.getIdentifier(), throwable.getCause()); } }, MoreExecutors.directExecutor()); futureMap.put(nodeConnectorLock, writeTxResultFuture); } } AddressObservationWriter(DataBroker dataService); void setTimestampUpdateInterval(long timestampUpdateInterval); void addAddress(MacAddress macAddress, IpAddress ipAddress, NodeConnectorRef nodeConnectorRef);  }
@Test public void addAddressTest() throws Exception { AddressObservationWriter addressObservationWriter = new AddressObservationWriter(dataService); addressObservationWriter.setTimestampUpdateInterval(20L); addressObservationWriter.addAddress(macAddress, ipAddress, realNcRef); verify(readTransaction, times(1)).read(any(LogicalDatastoreType.class), any(InstanceIdentifier.class)); verify(readTransaction, times(1)).close(); verify(writeTransaction, times(1)).merge(any(LogicalDatastoreType.class), any(InstanceIdentifier.class), any(AddressCapableNodeConnector.class)); verify(writeTransaction, times(1)).submit(); }