ActorUtils { public Optional<ActorRef> findLocalShard(final String shardName) { Object result = executeOperation(shardManager, new FindLocalShard(shardName, false)); if (result instanceof LocalShardFound) { LocalShardFound found = (LocalShardFound) result; LOG.debug("Local shard found {}", found.getPath()); return Optional.of(found.getPath()); } return Optional.empty(); } 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 testFindLocalShardWithShardFound() { final TestKit testKit = new TestKit(getSystem()); testKit.within(Duration.ofSeconds(1), () -> { ActorRef shardActorRef = getSystem().actorOf(Props.create(EchoActor.class)); ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(true, shardActorRef)); ActorUtils actorUtils = new ActorUtils(getSystem(), shardManagerActorRef, mock(ClusterWrapper.class), mock(Configuration.class)); Optional<ActorRef> out = actorUtils.findLocalShard("default"); assertEquals(shardActorRef, out.get()); testKit.expectNoMessage(); return null; }); }
@Test public void testFindLocalShardWithShardNotFound() { ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(false, null)); ActorUtils actorUtils = new ActorUtils(getSystem(), shardManagerActorRef, mock(ClusterWrapper.class), mock(Configuration.class)); Optional<ActorRef> out = actorUtils.findLocalShard("default"); assertFalse(out.isPresent()); }