@Test public void testRateIsRespected() throws Exception { int rate = 2000; if (connectionFactory instanceof HTTP2CServerConnectionFactory) { ((HTTP2CServerConnectionFactory)connectionFactory).setMaxConcurrentStreams(rate); } prepare(new TestHandler()); int iterations = 5 * rate; AtomicLong requests = new AtomicLong(); LoadGenerator loadGenerator = new LoadGenerator.Builder() .port(connector.getLocalPort()) .httpClientTransportBuilder(clientTransportBuilder) .iterationsPerThread(iterations) .resourceRate(rate) .socketAddressResolver(new SocketAddressResolver.Sync()) .requestListener(new Request.Listener.Adapter() { @Override public void onBegin(Request request) { requests.incrementAndGet(); } }) .build(); long start = System.nanoTime(); loadGenerator.begin().get(10 * rate, TimeUnit.MILLISECONDS); long elapsed = System.nanoTime() - start; long expected = TimeUnit.SECONDS.toNanos(iterations / rate); Assert.assertTrue(Math.abs(elapsed - expected) < expected / 10); Assert.assertEquals(iterations, requests.intValue()); }
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();  }