ActorUtils { public void broadcast(final Function<Short, Object> messageSupplier, final Class<?> messageClass) { for (final String shardName : configuration.getAllShardNames()) { Future<PrimaryShardInfo> primaryFuture = findPrimaryShardAsync(shardName); primaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() { @Override public void onComplete(final Throwable failure, final PrimaryShardInfo primaryShardInfo) { if (failure != null) { LOG.warn("broadcast failed to send message {} to shard {}", messageClass.getSimpleName(), shardName, failure); } else { Object message = messageSupplier.apply(primaryShardInfo.getPrimaryShardVersion()); primaryShardInfo.getPrimaryShardActor().tell(message, ActorRef.noSender()); } } }, getClientDispatcher()); } } ActorUtils(final ActorSystem actorSystem, final ActorRef shardManager,
            final ClusterWrapper clusterWrapper, final Configuration configuration);  ActorUtils(final ActorSystem actorSystem, final ActorRef shardManager,
            final ClusterWrapper clusterWrapper, final Configuration configuration,
            final DatastoreContext datastoreContext, final PrimaryShardInfoFutureCache primaryShardInfoCache); DatastoreContext getDatastoreContext(); ActorSystem getActorSystem(); ActorRef getShardManager(); ActorSelection actorSelection(final String actorPath); ActorSelection actorSelection(final ActorPath actorPath); void setSchemaContext(final EffectiveModelContext schemaContext); void setDatastoreContext(final DatastoreContextFactory contextFactory); EffectiveModelContext getSchemaContext(); Future<PrimaryShardInfo> findPrimaryShardAsync(final String shardName); Optional<ActorRef> findLocalShard(final String shardName); Future<ActorRef> findLocalShardAsync(final String shardName); @SuppressWarnings("checkstyle:IllegalCatch") Object executeOperation(final ActorRef actor, final Object message); @SuppressWarnings("checkstyle:IllegalCatch") Object executeOperation(final ActorSelection actor, final Object message); Future<Object> executeOperationAsync(final ActorRef actor, final Object message, final Timeout timeout); Future<Object> executeOperationAsync(final ActorSelection actor, final Object message,
            final Timeout timeout); Future<Object> executeOperationAsync(final ActorSelection actor, final Object message); void sendOperationAsync(final ActorSelection actor, final Object message); @SuppressWarnings("checkstyle:IllegalCatch") void shutdown(); ClusterWrapper getClusterWrapper(); MemberName getCurrentMemberName(); void broadcast(final Function<Short, Object> messageSupplier, final Class<?> messageClass); FiniteDuration getOperationDuration(); Timeout getOperationTimeout(); boolean isPathLocal(final String path); Timer getOperationTimer(final String operationName); Timer getOperationTimer(final String dataStoreType, final String operationName); String getDataStoreName(); double getTxCreationLimit(); void acquireTxCreationPermit(); Timeout getTransactionCommitOperationTimeout(); ExecutionContext getClientDispatcher(); String getNotificationDispatcherPath(); Configuration getConfiguration(); ShardStrategyFactory getShardStrategyFactory(); PrimaryShardInfoFutureCache getPrimaryShardInfoCache(); static final String BOUNDED_MAILBOX; static final String COMMIT; }
@Test public void testBroadcast() { ActorRef shardActorRef1 = getSystem().actorOf(MessageCollectorActor.props()); ActorRef shardActorRef2 = getSystem().actorOf(MessageCollectorActor.props()); TestActorRef<MockShardManager> shardManagerActorRef = TestActorRef.create(getSystem(), MockShardManager.props()); MockShardManager shardManagerActor = shardManagerActorRef.underlyingActor(); shardManagerActor.addFindPrimaryResp("shard1", new RemotePrimaryShardFound( shardActorRef1.path().toString(), DataStoreVersions.CURRENT_VERSION)); shardManagerActor.addFindPrimaryResp("shard2", new RemotePrimaryShardFound( shardActorRef2.path().toString(), DataStoreVersions.CURRENT_VERSION)); shardManagerActor.addFindPrimaryResp("shard3", new NoShardLeaderException("not found")); Configuration mockConfig = mock(Configuration.class); doReturn(Sets.newLinkedHashSet(Arrays.asList("shard1", "shard2", "shard3"))).when(mockConfig) .getAllShardNames(); ActorUtils actorUtils = new ActorUtils(getSystem(), shardManagerActorRef, mock(ClusterWrapper.class), mockConfig, DatastoreContext.newBuilder().shardInitializationTimeout(200, TimeUnit.MILLISECONDS).build(), new PrimaryShardInfoFutureCache()); actorUtils.broadcast(v -> new TestMessage(), TestMessage.class); MessageCollectorActor.expectFirstMatching(shardActorRef1, TestMessage.class); MessageCollectorActor.expectFirstMatching(shardActorRef2, TestMessage.class); }