JobNodeStorage { public void addDataListener(final CuratorCacheListener listener) { CuratorCache cache = (CuratorCache) regCenter.getRawCache("/" + jobName); cache.listenable().addListener(listener); } JobNodeStorage(final CoordinatorRegistryCenter regCenter, final String jobName); boolean isJobNodeExisted(final String node); boolean isJobRootNodeExisted(); String getJobNodeData(final String node); String getJobNodeDataDirectly(final String node); List<String> getJobNodeChildrenKeys(final String node); String getJobRootNodeData(); void createJobNodeIfNeeded(final String node); void removeJobNodeIfExisted(final String node); void fillJobNode(final String node, final Object value); void fillEphemeralJobNode(final String node, final Object value); void updateJobNode(final String node, final Object value); void replaceJobNode(final String node, final Object value); void replaceJobRootNode(final Object value); void executeInTransaction(final TransactionExecutionCallback callback); void executeInLeader(final String latchNode, final LeaderExecutionCallback callback); void addConnectionStateListener(final ConnectionStateListener listener); void addDataListener(final CuratorCacheListener listener); long getRegistryCenterTime();  }
@Test public void assertAddDataListener() { CuratorCache cache = mock(CuratorCache.class); @SuppressWarnings("unchecked") Listenable<CuratorCacheListener> listeners = mock(Listenable.class); CuratorCacheListener listener = mock(CuratorCacheListener.class); when(cache.listenable()).thenReturn(listeners); when(regCenter.getRawCache("/test_job")).thenReturn(cache); jobNodeStorage.addDataListener(listener); verify(listeners).addListener(listener); }