ModuleShardBackendResolver extends AbstractShardBackendResolver { @Override public CompletionStage<ShardBackendInfo> getBackendInfo(final Long cookie) { final ShardState existing = backends.get(cookie); if (existing != null) { return existing.getStage(); } final String shardName = shards.inverse().get(cookie); if (shardName == null) { LOG.warn("Failing request for non-existent cookie {}", cookie); throw new IllegalArgumentException("Cookie " + cookie + " does not have a shard assigned"); } LOG.debug("Resolving cookie {} to shard {}", cookie, shardName); final ShardState toInsert = resolveBackendInfo(shardName, cookie); final ShardState raced = backends.putIfAbsent(cookie, toInsert); if (raced != null) { LOG.debug("Race during insertion of state for cookie {} shard {}", cookie, shardName); return raced.getStage(); } final CompletionStage<ShardBackendInfo> stage = toInsert.getStage(); stage.whenComplete((info, failure) -> { if (failure != null) { LOG.debug("Resolution of cookie {} shard {} failed, removing state", cookie, shardName, failure); backends.remove(cookie, toInsert); flushCache(shardName); } }); return stage; } ModuleShardBackendResolver(final ClientIdentifier clientId, final ActorUtils actorUtils); @Override CompletionStage<ShardBackendInfo> getBackendInfo(final Long cookie); @Override CompletionStage<ShardBackendInfo> refreshBackendInfo(final Long cookie,
            final ShardBackendInfo staleInfo); @Override void close(); @Override String resolveCookieName(Long cookie);  }
@Test public void testGetBackendInfo() throws Exception { final CompletionStage<ShardBackendInfo> i = moduleShardBackendResolver.getBackendInfo(0L); contextProbe.expectMsgClass(ConnectClientRequest.class); final TestProbe backendProbe = new TestProbe(system, "backend"); final ConnectClientSuccess msg = new ConnectClientSuccess(CLIENT_ID, 0L, backendProbe.ref(), Collections.emptyList(), dataTree, 3); contextProbe.reply(msg); final CompletionStage<ShardBackendInfo> stage = moduleShardBackendResolver.getBackendInfo(0L); final ShardBackendInfo shardBackendInfo = TestUtils.getWithTimeout(stage.toCompletableFuture()); Assert.assertEquals(0L, shardBackendInfo.getCookie().longValue()); Assert.assertEquals(dataTree, shardBackendInfo.getDataTree().get()); Assert.assertEquals(DefaultShardStrategy.DEFAULT_SHARD, shardBackendInfo.getName()); }
@Test public void testGetBackendInfoFail() throws Exception { final CompletionStage<ShardBackendInfo> i = moduleShardBackendResolver.getBackendInfo(0L); final ConnectClientRequest req = contextProbe.expectMsgClass(ConnectClientRequest.class); final RuntimeException cause = new RuntimeException(); final ConnectClientFailure response = req.toRequestFailure(new RuntimeRequestException("fail", cause)); contextProbe.reply(response); final CompletionStage<ShardBackendInfo> stage = moduleShardBackendResolver.getBackendInfo(0L); final ExecutionException caught = TestUtils.assertOperationThrowsException(() -> TestUtils.getWithTimeout(stage.toCompletableFuture()), ExecutionException.class); Assert.assertEquals(cause, caught.getCause()); }