Prepare release

benchmarks-api/src/main/java/io/scalecube/benchmarks/BenchmarksState.java

@@ -319,27 +319,24 @@ public final <T> void runWithRampUp(
             int schedulerIndex = (int) ((rampUpIteration & Long.MAX_VALUE) % schedulers().size());
             Scheduler scheduler = schedulers().get(schedulerIndex);
 
-            return Flux
-                .range(0, Math.max(1, settings.injectorsPerRampUpInterval()))
-                .flatMap(iteration1 -> {
-                  // create tasks on selected scheduler
-                  Flux<T> setUpFactory = Flux.create((FluxSink<T> sink) -> {
-                    Flux<T> deferSetUp = Flux.defer(() -> setUp.apply(rampUpIteration, self));
-                    deferSetUp.subscribe(
-                        sink::next,
-                        ex -> {
-                          LOGGER.error("Exception occured on setUp at rampUpIteration: {}, "
-                              + "cause: {}, task won't start", rampUpIteration, ex);
-                          sink.complete();
-                        },
-                        sink::complete);
-                  });
-                  return setUpFactory
-                      .subscribeOn(scheduler)
-                      .map(setUpResult -> new BenchmarksTask<>(self, setUpResult, unitOfWork, cleanUp, scheduler))
-                      .doOnNext(scheduler::schedule)
-                      .flatMap(BenchmarksTask::completionMono);
-                });
+            // create tasks on selected scheduler
+            Flux<T> setUpFactory = Flux.create((FluxSink<T> sink) -> {
+              Flux<T> deferSetUp = Flux.defer(() -> setUp.apply(rampUpIteration, self));
+              deferSetUp.subscribe(
+                  sink::next,
+                  ex -> {
+                    LOGGER.error("Exception occured on setUp at rampUpIteration: {}, "
+                        + "cause: {}, task won't start", rampUpIteration, ex);
+                    sink.complete();
+                  },
+                  sink::complete);
+            });
+            return setUpFactory
+                .subscribeOn(scheduler)
+                .map(setUpResult -> new BenchmarksTask<>(self, setUpResult, unitOfWork, cleanUp, scheduler))
+                .doOnNext(scheduler::schedule)
+                .flatMap(BenchmarksTask::completionMono)
+                .repeat(Math.max(1, settings.injectorsPerRampUpInterval()));
           }, Integer.MAX_VALUE, Integer.MAX_VALUE)
           .blockLast();
     } finally {

benchmarks-api/src/main/java/io/scalecube/benchmarks/BenchmarksTask.java

@@ -81,17 +81,16 @@ public void run() {
     if (isScheduled()) { // executing
       long iter = iterationsCounter.incrementAndGet();
 
-      Flux
-          .range(0, Math.max(1, benchmarksState.settings.messagesPerExecutionInterval()))
-          .flatMap(iteration1 -> Flux.from(unitOfWork.apply(iter, setUpResult)))
-          .doOnError(ex -> LOGGER.warn("Exception occured on unitOfWork at iteration: {}, cause: {}", iter, ex))
-          .subscribe();
-
-      if (executionTaskInterval.isZero()) {
-        scheduler.schedule(this);
-      } else {
-        scheduler.schedule(this, executionTaskInterval.toMillis(), TimeUnit.MILLISECONDS);
-      }
+      Flux.from(unitOfWork.apply(iter, setUpResult))
+          .doOnError(ex -> LOGGER.warn("Exception occurred on unitOfWork at iteration: {}, cause: {}", iter, ex))
+          .repeat(Math.max(1, benchmarksState.settings.messagesPerExecutionInterval()))
+          .doFinally(signalType -> {
+            if (executionTaskInterval.isZero()) {
+              scheduler.schedule(this);
+            } else {
+              scheduler.schedule(this, executionTaskInterval.toMillis(), TimeUnit.MILLISECONDS);
+            }
+          }).subscribe();
       return;
     }
 

benchmarks-examples/src/main/java/io/scalecube/benchmarks/examples/RampUpSimpleTestRunner.java

@@ -4,48 +4,48 @@
 
 import reactor.core.publisher.Mono;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.concurrent.TimeUnit;
 
 public class RampUpSimpleTestRunner {
 
+  public static final Logger LOGGER = LoggerFactory.getLogger(RampUpSimpleTestRunner.class);
+
   /**
    * Runs example benchmark.
    *
    * @param args command line args
    */
   public static void main(String[] args) {
     BenchmarksSettings settings = BenchmarksSettings.from(args)
-        .injectors(1000)
-        .messageRate(10_000)
-        .rampUpDuration(Duration.ofSeconds(15))
-        .executionTaskDuration(Duration.ofSeconds(30))
+        .injectors(5)
+        .messageRate(5)
+        .rampUpDuration(Duration.ofSeconds(5))
+        .executionTaskDuration(Duration.ofSeconds(10))
         .consoleReporterEnabled(false)
         .durationUnit(TimeUnit.NANOSECONDS)
         .build();
 
-    System.out.println("Settings:");
-    System.out.println(settings);
-    System.out.println(LocalDateTime.now() + " Test started");
-
     new ExampleServiceBenchmarksState(settings).runWithRampUp(
         // set up
         (rampUpIteration, state) -> {
-          // System.out.println(LocalDateTime.now() + " User started: " + rampUpIteration);
+          LOGGER.info("User started: " + rampUpIteration);
           return Mono.just(rampUpIteration);
         },
 
         // job
         state -> (iteration, userId) -> {
-          // System.out.println(LocalDateTime.now() + " User: " + userId + " | iteration: " + iteration);
-          ;
+          LOGGER.info("User: " + userId + " | iteration: " + iteration);
           return Mono.fromRunnable(RampUpSimpleTestRunner::heavy);
         },
 
         // teardown
         (state, userId) -> {
-          // System.out.println(LocalDateTime.now() + " User done:" + userId);
+          LOGGER.info("User done:" + userId);
           return Mono.empty();
         });
     System.out.println(LocalDateTime.now() + " Test over");