DistributedShardedDOMDataTree implements DOMDataTreeService, DOMDataTreeShardingService,
        DistributedShardFactory { @Override public CompletionStage<DistributedShardRegistration> createDistributedShard( final DOMDataTreeIdentifier prefix, final Collection<MemberName> replicaMembers) throws DOMDataTreeShardingConflictException { synchronized (shards) { final DOMDataTreePrefixTableEntry<DOMDataTreeShardRegistration<DOMDataTreeShard>> lookup = shards.lookup(prefix); if (lookup != null && lookup.getValue().getPrefix().equals(prefix)) { throw new DOMDataTreeShardingConflictException( "Prefix " + prefix + " is already occupied by another shard."); } } final PrefixedShardConfigWriter writer = writerMap.get(prefix.getDatastoreType()); final ListenableFuture<Void> writeFuture = writer.writeConfig(prefix.getRootIdentifier(), replicaMembers); final Promise<DistributedShardRegistration> shardRegistrationPromise = akka.dispatch.Futures.promise(); Futures.addCallback(writeFuture, new FutureCallback<Void>() { @Override public void onSuccess(final Void result) { final Future<Object> ask = Patterns.ask(shardedDataTreeActor, new LookupPrefixShard(prefix), SHARD_FUTURE_TIMEOUT); shardRegistrationPromise.completeWith(ask.transform( new Mapper<Object, DistributedShardRegistration>() { @Override public DistributedShardRegistration apply(final Object parameter) { return new DistributedShardRegistrationImpl( prefix, shardedDataTreeActor, DistributedShardedDOMDataTree.this); } }, new Mapper<Throwable, Throwable>() { @Override public Throwable apply(final Throwable throwable) { return new DOMDataTreeShardCreationFailedException( "Unable to create a cds shard.", throwable); } }, actorSystem.dispatcher())); } @Override public void onFailure(final Throwable throwable) { shardRegistrationPromise.failure( new DOMDataTreeShardCreationFailedException("Unable to create a cds shard.", throwable)); } }, MoreExecutors.directExecutor()); return FutureConverters.toJava(shardRegistrationPromise.future()); } DistributedShardedDOMDataTree(final ActorSystemProvider actorSystemProvider,
                                         final DistributedDataStoreInterface distributedOperDatastore,
                                         final DistributedDataStoreInterface distributedConfigDatastore); void init(); @Override ListenerRegistration<T> registerListener(
            final T listener, final Collection<DOMDataTreeIdentifier> subtrees,
            final boolean allowRxMerges, final Collection<DOMDataTreeProducer> producers); @Override ClassToInstanceMap<DOMDataTreeServiceExtension> getExtensions(); @Override DOMDataTreeProducer createProducer(final Collection<DOMDataTreeIdentifier> subtrees); @Override CompletionStage<DistributedShardRegistration> createDistributedShard(
            final DOMDataTreeIdentifier prefix, final Collection<MemberName> replicaMembers); @Override ListenerRegistration<T> registerDataTreeShard(
            final DOMDataTreeIdentifier prefix, final T shard, final DOMDataTreeProducer producer);  }
@Test public void testMultipleRegistrationsAtOnePrefix() throws Exception { initEmptyDatastores(); for (int i = 0; i < 10; i++) { LOG.debug("Round {}", i); final DistributedShardRegistration reg1 = waitOnAsyncTask(leaderShardFactory.createDistributedShard( TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION); leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorUtils(), ClusterUtils.getCleanShardName(TestModel.TEST_PATH)); assertNotNull(findLocalShard(leaderDistributedDataStore.getActorUtils(), ClusterUtils.getCleanShardName(TestModel.TEST_PATH))); waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION); waitUntilShardIsDown(leaderDistributedDataStore.getActorUtils(), ClusterUtils.getCleanShardName(TestModel.TEST_PATH)); } }