@Test public void testRateRampUp() throws Exception { prepare(new TestHandler()); int rate = 10; long ramp = 5; AtomicLong requests = new AtomicLong(); LoadGenerator loadGenerator = new LoadGenerator.Builder() .port(connector.getLocalPort()) .httpClientTransportBuilder(clientTransportBuilder) .resourceRate(rate) .rateRampUpPeriod(ramp) .requestListener(new Request.Listener.Adapter() { @Override public void onBegin(Request request) { requests.incrementAndGet(); } }) .runFor(ramp, TimeUnit.SECONDS) .build(); loadGenerator.begin().get(); long expected = rate * ramp / 2; Assert.assertTrue(expected - 1 <= requests.get()); Assert.assertTrue(requests.get() <= expected + 1); }
public CompletableFuture<Void> begin() { if (logger.isDebugEnabled()) { logger.debug("generating load, {}", config); } return proceed().thenCompose(x -> { CompletableFuture[] futures = new CompletableFuture[config.getThreads()]; for (int i = 0; i < futures.length; ++i) { futures[i] = CompletableFuture.supplyAsync(this::process, executorService).thenCompose(Function.identity()); } return CompletableFuture.allOf(futures); }).thenComposeAsync(x -> halt(), executorService); }
LoadGenerator extends ContainerLifeCycle { public CompletableFuture<Void> begin() { if (logger.isDebugEnabled()) { logger.debug("generating load, {}", config); } return proceed().thenCompose(x -> { CompletableFuture[] futures = new CompletableFuture[config.getThreads()]; for (int i = 0; i < futures.length; ++i) { futures[i] = CompletableFuture.supplyAsync(this::process, executorService).thenCompose(Function.identity()); } return CompletableFuture.allOf(futures); }).thenComposeAsync(x -> halt(), executorService); } }
LoadGenerator extends ContainerLifeCycle { public CompletableFuture<Void> begin() { if (logger.isDebugEnabled()) { logger.debug("generating load, {}", config); } return proceed().thenCompose(x -> { CompletableFuture[] futures = new CompletableFuture[config.getThreads()]; for (int i = 0; i < futures.length; ++i) { futures[i] = CompletableFuture.supplyAsync(this::process, executorService).thenCompose(Function.identity()); } return CompletableFuture.allOf(futures); }).thenComposeAsync(x -> halt(), executorService); } private  LoadGenerator(Config config); }
LoadGenerator extends ContainerLifeCycle { public CompletableFuture<Void> begin() { if (logger.isDebugEnabled()) { logger.debug("generating load, {}", config); } return proceed().thenCompose(x -> { CompletableFuture[] futures = new CompletableFuture[config.getThreads()]; for (int i = 0; i < futures.length; ++i) { futures[i] = CompletableFuture.supplyAsync(this::process, executorService).thenCompose(Function.identity()); } return CompletableFuture.allOf(futures); }).thenComposeAsync(x -> halt(), executorService); } private  LoadGenerator(Config config); Config getConfig(); CompletableFuture<Void> begin(); @ManagedOperation(value = "Interrupts this LoadGenerator", impact = "ACTION") void interrupt(); }
LoadGenerator extends ContainerLifeCycle { public CompletableFuture<Void> begin() { if (logger.isDebugEnabled()) { logger.debug("generating load, {}", config); } return proceed().thenCompose(x -> { CompletableFuture[] futures = new CompletableFuture[config.getThreads()]; for (int i = 0; i < futures.length; ++i) { futures[i] = CompletableFuture.supplyAsync(this::process, executorService).thenCompose(Function.identity()); } return CompletableFuture.allOf(futures); }).thenComposeAsync(x -> halt(), executorService); } private  LoadGenerator(Config config); Config getConfig(); CompletableFuture<Void> begin(); @ManagedOperation(value = "Interrupts this LoadGenerator", impact = "ACTION") void interrupt();  }