EventSourceTopic implements DataTreeChangeListener<Node>, AutoCloseable { public void notifyNode(final InstanceIdentifier<?> nodeId) { LOG.debug("Notify node: {}", nodeId); try { final RpcResult<JoinTopicOutput> rpcResultJoinTopic = sourceService.joinTopic(getJoinTopicInputArgument(nodeId)).get(); if (!rpcResultJoinTopic.isSuccessful()) { for (final RpcError err : rpcResultJoinTopic.getErrors()) { LOG.error("Can not join topic: [{}] on node: [{}]. Error: {}", getTopicId().getValue(), nodeId.toString(), err.toString()); } } else { joinedEventSources.add(nodeId); } } catch (InterruptedException | ExecutionException e) { LOG.error("Could not invoke join topic for node {}", nodeId); } } private  EventSourceTopic(final NotificationPattern notificationPattern, final String nodeIdRegexPattern,
            final EventSourceService sourceService); static EventSourceTopic create(final NotificationPattern notificationPattern,
            final String nodeIdRegexPattern, final EventSourceTopology eventSourceTopology); TopicId getTopicId(); @Override void onDataTreeChanged(final Collection<DataTreeModification<Node>> changes); void notifyNode(final InstanceIdentifier<?> nodeId); Pattern getNodeIdRegexPattern(); @Override void close();  }
@Test public void notifyNodeTest() { InstanceIdentifier instanceIdentifierMock = mock(InstanceIdentifier.class); eventSourceTopic.notifyNode(instanceIdentifierMock); verify(eventSourceServiceMock, times(1)).joinTopic(any(JoinTopicInput.class)); }