@Test public void testDocExample() throws Exception { CuratorOp op1 = client.transactionOp().create().forPath("/parent"); CuratorOp op2 = client.transactionOp().create().forPath("/parent/one"); CuratorOp op3 = client.transactionOp().create().forPath("/parent/two"); CuratorOp op4 = client.transactionOp().create().forPath("/parent/three"); CuratorOp op5 = client.transactionOp().create().forPath("/main", "hey".getBytes()); Migration initialMigration = () -> Arrays.asList(op1, op2, op3, op4, op5); MigrationSet migrationSet = MigrationSet.build("main", Collections.singletonList(initialMigration)); complete(manager.migrate(migrationSet)); Assert.assertNotNull(client.unwrap().checkExists().forPath("/parent/three")); Assert.assertEquals(client.unwrap().getData().forPath("/main"), "hey".getBytes()); CuratorOp newOp1 = client.transactionOp().create().forPath("/new"); CuratorOp newOp2 = client.transactionOp().delete().forPath("/main"); Migration newMigration = () -> Arrays.asList(newOp1, newOp2); migrationSet = MigrationSet.build("main", Arrays.asList(initialMigration, newMigration)); complete(manager.migrate(migrationSet)); Assert.assertNull(client.unwrap().checkExists().forPath("/main")); }
public CompletionStage<Void> migrate(MigrationSet set) { InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id())); CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor); return lockStage.thenCompose(__ -> runMigrationInLock(lock, set)); }
MigrationManager { public CompletionStage<Void> migrate(MigrationSet set) { InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id())); CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor); return lockStage.thenCompose(__ -> runMigrationInLock(lock, set)); } }
MigrationManager { public CompletionStage<Void> migrate(MigrationSet set) { InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id())); CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor); return lockStage.thenCompose(__ -> runMigrationInLock(lock, set)); } MigrationManager(AsyncCuratorFramework client, String lockPath, String metaDataPath, Executor executor, Duration lockMax); }
MigrationManager { public CompletionStage<Void> migrate(MigrationSet set) { InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id())); CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor); return lockStage.thenCompose(__ -> runMigrationInLock(lock, set)); } MigrationManager(AsyncCuratorFramework client, String lockPath, String metaDataPath, Executor executor, Duration lockMax); CompletionStage<Void> migrate(MigrationSet set); }
MigrationManager { public CompletionStage<Void> migrate(MigrationSet set) { InterProcessLock lock = new InterProcessSemaphoreMutex(client.unwrap(), ZKPaths.makePath(lockPath, set.id())); CompletionStage<Void> lockStage = lockAsync(lock, lockMax.toMillis(), TimeUnit.MILLISECONDS, executor); return lockStage.thenCompose(__ -> runMigrationInLock(lock, set)); } MigrationManager(AsyncCuratorFramework client, String lockPath, String metaDataPath, Executor executor, Duration lockMax); CompletionStage<Void> migrate(MigrationSet set);  }