JobNodeStorage { public void executeInTransaction(final TransactionExecutionCallback callback) { try { CuratorTransactionFinal curatorTransactionFinal = getClient().inTransaction().check().forPath("/").and(); callback.execute(curatorTransactionFinal); curatorTransactionFinal.commit(); } catch (final Exception ex) { RegExceptionHandler.handleException(ex); } } JobNodeStorage(final CoordinatorRegistryCenter coordinatorRegistryCenter, final JobConfiguration jobConfiguration); boolean isJobNodeExisted(final String node); String getJobNodeData(final String node); String getJobNodeDataDirectly(final String node); List<String> getJobNodeChildrenKeys(final String node); void createJobNodeIfNeeded(final String node); void removeJobNodeIfExisted(final String node); void fillJobNodeIfNullOrOverwrite(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 executeInTransaction(final TransactionExecutionCallback callback); void executeInLeader(final String latchNode, final LeaderExecutionCallback callback); void addConnectionStateListener(final ConnectionStateListener listener); void addDataListener(final TreeCacheListener listener); long getRegistryCenterTime();  }
@Test public void assertExecuteInTransactionSuccess() throws Exception { CuratorFramework client = mock(CuratorFramework.class); CuratorTransaction curatorTransaction = mock(CuratorTransaction.class); TransactionCheckBuilder transactionCheckBuilder = mock(TransactionCheckBuilder.class); CuratorTransactionBridge curatorTransactionBridge = mock(CuratorTransactionBridge.class); CuratorTransactionFinal curatorTransactionFinal = mock(CuratorTransactionFinal.class); when(coordinatorRegistryCenter.getRawClient()).thenReturn(client); when(client.inTransaction()).thenReturn(curatorTransaction); when(curatorTransaction.check()).thenReturn(transactionCheckBuilder); when(transactionCheckBuilder.forPath("/")).thenReturn(curatorTransactionBridge); when(curatorTransactionBridge.and()).thenReturn(curatorTransactionFinal); TransactionCreateBuilder transactionCreateBuilder = mock(TransactionCreateBuilder.class); when(curatorTransactionFinal.create()).thenReturn(transactionCreateBuilder); when(transactionCreateBuilder.forPath("/test_transaction")).thenReturn(curatorTransactionBridge); when(curatorTransactionBridge.and()).thenReturn(curatorTransactionFinal); jobNodeStorage.executeInTransaction(new TransactionExecutionCallback() { @Override public void execute(final CuratorTransactionFinal curatorTransactionFinal) throws Exception { curatorTransactionFinal.create().forPath("/test_transaction").and(); } }); verify(coordinatorRegistryCenter).getRawClient(); verify(client).inTransaction(); verify(curatorTransaction).check(); verify(transactionCheckBuilder).forPath("/"); verify(curatorTransactionBridge, times(2)).and(); verify(curatorTransactionFinal).create(); verify(transactionCreateBuilder).forPath("/test_transaction"); verify(curatorTransactionFinal).commit(); }
@Test(expected = RuntimeException.class) public void assertExecuteInTransactionFailure() throws Exception { CuratorFramework client = mock(CuratorFramework.class); CuratorTransaction curatorTransaction = mock(CuratorTransaction.class); TransactionCheckBuilder transactionCheckBuilder = mock(TransactionCheckBuilder.class); CuratorTransactionBridge curatorTransactionBridge = mock(CuratorTransactionBridge.class); CuratorTransactionFinal curatorTransactionFinal = mock(CuratorTransactionFinal.class); when(coordinatorRegistryCenter.getRawClient()).thenReturn(client); when(client.inTransaction()).thenReturn(curatorTransaction); when(curatorTransaction.check()).thenReturn(transactionCheckBuilder); when(transactionCheckBuilder.forPath("/")).thenReturn(curatorTransactionBridge); when(curatorTransactionBridge.and()).thenReturn(curatorTransactionFinal); TransactionCreateBuilder transactionCreateBuilder = mock(TransactionCreateBuilder.class); when(curatorTransactionFinal.create()).thenReturn(transactionCreateBuilder); when(transactionCreateBuilder.forPath("/test_transaction")).thenReturn(curatorTransactionBridge); when(curatorTransactionBridge.and()).thenThrow(new RuntimeException()); jobNodeStorage.executeInTransaction(new TransactionExecutionCallback() { @Override public void execute(final CuratorTransactionFinal curatorTransactionFinal) throws Exception { curatorTransactionFinal.create().forPath("/test_transaction").and(); } }); verify(coordinatorRegistryCenter).getRawClient(); verify(client).inTransaction(); verify(curatorTransaction).check(); verify(transactionCheckBuilder).forPath("/"); verify(curatorTransactionBridge, times(2)).and(); verify(curatorTransactionFinal).create(); verify(transactionCreateBuilder).forPath("/test_transaction"); verify(curatorTransactionFinal, times(0)).commit(); }