DataTreeChangeListenerProxy extends AbstractListenerRegistration<T> { void init(final String shardName) { Future<ActorRef> findFuture = actorUtils.findLocalShardAsync(shardName); findFuture.onComplete(new OnComplete<ActorRef>() { @Override public void onComplete(final Throwable failure, final ActorRef shard) { if (failure instanceof LocalShardNotFoundException) { LOG.debug("{}: No local shard found for {} - DataTreeChangeListener {} at path {} " + "cannot be registered", logContext(), shardName, getInstance(), registeredPath); } else if (failure != null) { LOG.error("{}: Failed to find local shard {} - DataTreeChangeListener {} at path {} " + "cannot be registered", logContext(), shardName, getInstance(), registeredPath, failure); } else { doRegistration(shard); } } }, actorUtils.getClientDispatcher()); } DataTreeChangeListenerProxy(final ActorUtils actorUtils, final T listener,
            final YangInstanceIdentifier registeredPath);   }
@Test(timeout = 10000) public void testSuccessfulRegistrationForClusteredListener() { final TestKit kit = new TestKit(getSystem()); ActorUtils actorUtils = new ActorUtils(getSystem(), kit.getRef(), mock(ClusterWrapper.class), mock(Configuration.class)); ClusteredDOMDataTreeChangeListener mockClusteredListener = mock( ClusteredDOMDataTreeChangeListener.class); final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME); final DataTreeChangeListenerProxy<ClusteredDOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>(actorUtils, mockClusteredListener, path); new Thread(() -> proxy.init("shard-1")).start(); Duration timeout = Duration.ofSeconds(5); FindLocalShard findLocalShard = kit.expectMsgClass(timeout, FindLocalShard.class); assertEquals("getShardName", "shard-1", findLocalShard.getShardName()); kit.reply(new LocalShardFound(kit.getRef())); RegisterDataTreeChangeListener registerMsg = kit.expectMsgClass(timeout, RegisterDataTreeChangeListener.class); assertEquals("getPath", path, registerMsg.getPath()); assertTrue("isRegisterOnAllInstances", registerMsg.isRegisterOnAllInstances()); proxy.close(); }
@Test(timeout = 10000) public void testLocalShardNotFound() { final TestKit kit = new TestKit(getSystem()); ActorUtils actorUtils = new ActorUtils(getSystem(), kit.getRef(), mock(ClusterWrapper.class), mock(Configuration.class)); final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME); final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>( actorUtils, mockListener, path); new Thread(() -> proxy.init("shard-1")).start(); Duration timeout = Duration.ofSeconds(5); FindLocalShard findLocalShard = kit.expectMsgClass(timeout, FindLocalShard.class); assertEquals("getShardName", "shard-1", findLocalShard.getShardName()); kit.reply(new LocalShardNotFound("shard-1")); kit.expectNoMessage(Duration.ofSeconds(1)); proxy.close(); }
@Test(timeout = 10000) public void testLocalShardNotInitialized() { final TestKit kit = new TestKit(getSystem()); ActorUtils actorUtils = new ActorUtils(getSystem(), kit.getRef(), mock(ClusterWrapper.class), mock(Configuration.class)); final YangInstanceIdentifier path = YangInstanceIdentifier.of(TestModel.TEST_QNAME); final DataTreeChangeListenerProxy<DOMDataTreeChangeListener> proxy = new DataTreeChangeListenerProxy<>( actorUtils, mockListener, path); new Thread(() -> proxy.init("shard-1")).start(); Duration timeout = Duration.ofSeconds(5); FindLocalShard findLocalShard = kit.expectMsgClass(timeout, FindLocalShard.class); assertEquals("getShardName", "shard-1", findLocalShard.getShardName()); kit.reply(new NotInitializedException("not initialized")); kit.within(Duration.ofSeconds(1), () -> { kit.expectNoMessage(); return null; }); proxy.close(); }