BatchingCallable extends UnaryCallable<RequestT, ResponseT> { @Override public ApiFuture<ResponseT> futureCall(RequestT request, ApiCallContext context) { if (batcherFactory.getBatchingSettings().getIsEnabled()) { BatchedFuture<ResponseT> result = BatchedFuture.<ResponseT>create(); UnaryCallable<RequestT, ResponseT> unaryCallable = callable.withDefaultCallContext(context); Batch<RequestT, ResponseT> batchableMessage = new Batch<RequestT, ResponseT>(batchingDescriptor, request, unaryCallable, result); PartitionKey partitionKey = batchingDescriptor.getBatchPartitionKey(request); ThresholdBatcher<Batch<RequestT, ResponseT>> batcher = batcherFactory.getPushingBatcher(partitionKey); try { batcher.add(batchableMessage); return result; } catch (FlowControlException e) { throw FlowControlRuntimeException.fromFlowControlException(e); } } else { return callable.futureCall(request, context); } } BatchingCallable(
      UnaryCallable<RequestT, ResponseT> callable,
      BatchingDescriptor<RequestT, ResponseT> batchingDescriptor,
      BatcherFactory<RequestT, ResponseT> batcherFactory); @Override ApiFuture<ResponseT> futureCall(RequestT request, ApiCallContext context);  }
@Test public void testBatchedCall() throws Exception { BatchingSettings batchingSettings = BatchingSettings.newBuilder() .setDelayThreshold(Duration.ofSeconds(10)) .setElementCountThreshold(2L) .setRequestByteThreshold(1000L) .build(); FlowControlSettings flowControlSettings = FlowControlSettings.newBuilder() .setLimitExceededBehavior(LimitExceededBehavior.Ignore) .build(); FlowController flowController = new FlowController(flowControlSettings); BatcherFactory<LabeledIntList, List<Integer>> batcherFactory = new BatcherFactory<>( new SquarerBatchingDescriptor(), batchingSettings, batchingExecutor, flowController); BatchingCallable<LabeledIntList, List<Integer>> batchingCallable = new BatchingCallable<>( FakeBatchableApi.callLabeledIntSquarer, FakeBatchableApi.SQUARER_BATCHING_DESC, batcherFactory); LabeledIntList request1 = new LabeledIntList("label", 2); ApiFuture<List<Integer>> future1 = batchingCallable.futureCall(request1, FakeCallContext.createDefault()); Truth.assertThat(future1.isDone()).isFalse(); LabeledIntList request2 = new LabeledIntList("label", 3); ApiFuture<List<Integer>> future2 = batchingCallable.futureCall(request2, FakeCallContext.createDefault()); List<Integer> response1 = future1.get(); List<Integer> response2 = future2.get(); Truth.assertThat(response1.size()).isEqualTo(1); Truth.assertThat(response1.get(0)).isEqualTo(2 * 2); Truth.assertThat(response2.size()).isEqualTo(1); Truth.assertThat(response2.get(0)).isEqualTo(3 * 3); }
@Test public void testUnbatchedCall() throws Exception { BatchingSettings batchingSettings = BatchingSettings.newBuilder().setIsEnabled(false).build(); FlowControlSettings flowControlSettings = FlowControlSettings.newBuilder() .setLimitExceededBehavior(LimitExceededBehavior.Ignore) .build(); FlowController flowController = new FlowController(flowControlSettings); BatcherFactory<LabeledIntList, List<Integer>> batcherFactory = new BatcherFactory<>( new SquarerBatchingDescriptor(), batchingSettings, batchingExecutor, flowController); BatchingCallable<LabeledIntList, List<Integer>> batchingCallable = new BatchingCallable<>( FakeBatchableApi.callLabeledIntSquarer, FakeBatchableApi.SQUARER_BATCHING_DESC, batcherFactory); LabeledIntList request1 = new LabeledIntList("label", 2); ApiFuture<List<Integer>> future1 = batchingCallable.futureCall(request1, FakeCallContext.createDefault()); List<Integer> response1 = future1.get(); Truth.assertThat(response1.size()).isEqualTo(1); Truth.assertThat(response1.get(0)).isEqualTo(2 * 2); }