FlowController { public void reserve(long elements, long bytes) throws FlowControlException { Preconditions.checkArgument(elements >= 0); Preconditions.checkArgument(bytes >= 0); if (outstandingElementCount != null) { if (!outstandingElementCount.acquire(elements)) { throw new MaxOutstandingElementCountReachedException(maxOutstandingElementCount); } } if (outstandingByteCount != null) { long permitsToDraw = Math.min(bytes, maxOutstandingRequestBytes); if (!outstandingByteCount.acquire(permitsToDraw)) { if (outstandingElementCount != null) { outstandingElementCount.release(elements); } throw new MaxOutstandingRequestBytesReachedException(maxOutstandingRequestBytes); } } } FlowController(FlowControlSettings settings); void reserve(long elements, long bytes); void release(long elements, long bytes);  }
@Test public void testInvalidArguments() throws Exception { FlowController flowController = new FlowController( FlowControlSettings.newBuilder() .setMaxOutstandingElementCount(10L) .setMaxOutstandingRequestBytes(10L) .setLimitExceededBehavior(LimitExceededBehavior.Block) .build()); flowController.reserve(0, 0); try { flowController.reserve(-1, 1); fail("Must have thrown an illegal argument error"); } catch (IllegalArgumentException expected) { } try { flowController.reserve(1, -1); fail("Must have thrown an illegal argument error"); } catch (IllegalArgumentException expected) { } }
@Test public void testRestoreAfterFail() throws FlowController.FlowControlException { FlowController flowController = new FlowController( FlowControlSettings.newBuilder() .setMaxOutstandingElementCount(2L) .setMaxOutstandingRequestBytes(1L) .setLimitExceededBehavior(LimitExceededBehavior.ThrowException) .build()); flowController.reserve(1, 1); try { flowController.reserve(1, 1); throw new IllegalStateException("flowController should not have any bytes left"); } catch (FlowController.MaxOutstandingRequestBytesReachedException e) { } flowController.reserve(1, 0); }