package io.scalecube.gateway.benchmarks;

import io.scalecube.services.api.ServiceMessage;
import java.util.concurrent.Callable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

public class BenchmarksServiceImpl implements BenchmarksService {

  @Override
  public Mono<ServiceMessage> one(ServiceMessage message) {
    Callable<ServiceMessage> callable =
        () -> {
          long value = System.currentTimeMillis();
          return ServiceMessage.from(message)
              .header(SERVICE_RECV_TIME, value)
              .header(SERVICE_SEND_TIME, value)
              .data("hello")
              .build();
        };
    return Mono.fromCallable(callable).subscribeOn(Schedulers.parallel());
  }

  @Override
  public Mono<ServiceMessage> failure(ServiceMessage message) {
    return Mono.defer(() -> Mono.error(new RuntimeException("General failure")));
  }

  @Override
  public Flux<ServiceMessage> infiniteStream(ServiceMessage message) {
    return Flux.defer(
        () -> {
          Callable<ServiceMessage> callable =
              () ->
                  ServiceMessage.builder()
                      .qualifier(message.qualifier())
                      .header(SERVICE_SEND_TIME, Long.toString(System.currentTimeMillis()))
                      .build();
          return Mono.fromCallable(callable).subscribeOn(Schedulers.parallel()).repeat();
        });
  }
}

@@ -1019,53 +1019,8 @@
 ) %7B%0A
-    return Flux.defer(%0A        () -%3E %7B%0A      
     
@@ -1067,26 +1067,14 @@
     
- 
-     
 () -%3E%0A
-      
     
@@ -1100,76 +1100,22 @@
 age.
-builder()%0A                      .qualifier(message.qualifier())%0A    
+from(message)%0A
     
@@ -1118,34 +1118,32 @@
                 
-  
 .header(SERVICE_
@@ -1157,22 +1157,8 @@
 ME, 
-Long.toString(
 Syst
@@ -1180,24 +1180,17 @@
 illis())
-)%0A      
+%0A
         
@@ -1203,30 +1203,24 @@
   .build();%0A
-      
     return M
@@ -1237,32 +1237,41 @@
 llable(callable)
+%0A        
 .subscribeOn(Sch
@@ -1289,16 +1289,25 @@
 allel())
+%0A        
 .repeat(
@@ -1307,17 +1307,16 @@
 repeat()
-;
 %0A       
@@ -1316,17 +1316,36 @@
         
-%7D
+.onBackpressureDrop(
 );%0A  %7D%0A%7D

Set stream scenario be in sync with stream scenario from services