@Test public void testMultipleThreads() throws Exception { prepare(new TestHandler()); Set<String> threads = Collections.newSetFromMap(new ConcurrentHashMap<>()); LoadGenerator loadGenerator = new LoadGenerator.Builder() .port(connector.getLocalPort()) .httpClientTransportBuilder(clientTransportBuilder) .threads(2) .iterationsPerThread(1) .requestListener(new Request.Listener.Adapter() { @Override public void onBegin(Request request) { threads.add(Thread.currentThread().getName()); } }) .build(); loadGenerator.begin().get(5, TimeUnit.SECONDS); Assert.assertEquals(2, threads.size()); }
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();  }