use switchOnFirst replacing HeadAndTail

pom.xml

@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.scalecube</groupId>
     <artifactId>scalecube-parent-pom</artifactId>
-    <version>0.0.17</version>
+    <version>0.0.19</version>
   </parent>
   <packaging>pom</packaging>
 
@@ -23,20 +23,22 @@
 
   <properties>
     <jackson.version>2.9.8</jackson.version>
-    <scalecube-cluster.version>2.2.3</scalecube-cluster.version>
+    <scalecube-cluster.version>2.2.5</scalecube-cluster.version>
     <scalecube-benchmarks.version>1.2.2</scalecube-benchmarks.version>
-    <scalecube-config.version>0.3.4</scalecube-config.version>
-    <reactivestreams.version>1.0.2</reactivestreams.version>
+    <scalecube-config.version>0.3.6</scalecube-config.version>
+    <reactor.version>Californium-SR5</reactor.version>
+    <rsocket.version>0.11.17</rsocket.version>
     <metrics.version>3.1.2</metrics.version>
-    <rsocket.version>0.11.16</rsocket.version>
     <protostuff.version>1.6.0</protostuff.version>
-    <netty.version>4.1.31.Final</netty.version>
+    <netty.version>4.1.33.Final</netty.version>
     <slf4j.version>1.7.7</slf4j.version>
     <log4j.version>2.11.0</log4j.version>
     <disruptor.version>3.4.2</disruptor.version>
     <jsr305.version>3.0.2</jsr305.version>
     <jctools.version>2.1.2</jctools.version>
-    <hamcrest-all.version>1.3</hamcrest-all.version>
+    <junit.version>5.1.1</junit.version>
+    <mockito.version>2.24.5</mockito.version>
+    <hamcrest.version>1.3</hamcrest.version>
   </properties>
 
   <modules>
@@ -75,13 +77,13 @@
         <version>${scalecube-config.version}</version>
       </dependency>
 
-
-
-      <!-- Reactive Streams -->
+      <!-- Reactor -->
       <dependency>
-        <groupId>org.reactivestreams</groupId>
-        <artifactId>reactive-streams</artifactId>
-        <version>${reactivestreams.version}</version>
+        <groupId>io.projectreactor</groupId>
+        <artifactId>reactor-bom</artifactId>
+        <version>${reactor.version}</version>
+        <type>pom</type>
+        <scope>import</scope>
       </dependency>
 
       <!-- Logging -->
@@ -226,14 +228,6 @@
         <artifactId>jctools-core</artifactId>
         <version>${jctools.version}</version>
       </dependency>
-
-      <!-- Test dependencies -->
-      <dependency>
-        <groupId>org.hamcrest</groupId>
-        <artifactId>hamcrest-all</artifactId>
-        <version>${hamcrest-all.version}</version>
-        <scope>test</scope>
-      </dependency>
     </dependencies>
   </dependencyManagement>
 
@@ -242,26 +236,31 @@
     <dependency>
       <groupId>org.junit.jupiter</groupId>
       <artifactId>junit-jupiter-engine</artifactId>
+      <version>${junit.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.junit.jupiter</groupId>
       <artifactId>junit-jupiter-params</artifactId>
+      <version>${junit.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-junit-jupiter</artifactId>
+      <version>${mockito.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.hamcrest</groupId>
       <artifactId>hamcrest-all</artifactId>
+      <version>${hamcrest.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.hamcrest</groupId>
       <artifactId>hamcrest-core</artifactId>
+      <version>${hamcrest.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>

services-api/pom.xml

@@ -30,17 +30,6 @@
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
     </dependency>
-
-    <dependency>
-      <groupId>org.hamcrest</groupId>
-      <artifactId>hamcrest-all</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>io.projectreactor</groupId>
-      <artifactId>reactor-test</artifactId>
-      <scope>test</scope>
-    </dependency>
   </dependencies>
 
 </project>

services-gateway-http/src/main/java/io/scalecube/services/gateway/http/HttpGatewayAcceptor.java

@@ -108,7 +108,7 @@ private Mono<Void> error(HttpServerResponse httpResponse, ServiceMessage respons
     ByteBuf content =
         response.hasData(ErrorData.class)
             ? encodeData(response.data(), response.dataFormatOrDefault())
-            : response.data();
+            : ((ByteBuf) response.data()).retain();
 
     return httpResponse.status(status).sendObject(content).then();
   }
@@ -120,7 +120,7 @@ private Mono<Void> noContent(HttpServerResponse httpResponse) {
   private Mono<Void> ok(HttpServerResponse httpResponse, ServiceMessage response) {
     ByteBuf content =
         response.hasData(ByteBuf.class)
-            ? response.data()
+            ? ((ByteBuf) response.data()).retain()
             : encodeData(response.data(), response.dataFormatOrDefault());
 
     return httpResponse.status(OK).sendObject(content).then();

services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServiceAcceptor.java

@@ -6,7 +6,6 @@
 import io.rsocket.RSocket;
 import io.rsocket.SocketAcceptor;
 import io.rsocket.util.ByteBufPayload;
-import io.scalecube.services.HeadAndTail;
 import io.scalecube.services.api.ServiceMessage;
 import io.scalecube.services.exceptions.BadRequestException;
 import io.scalecube.services.exceptions.ServiceException;
@@ -72,14 +71,20 @@ public Flux<Payload> requestStream(Payload payload) {
 
     @Override
     public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
-      return Flux.from(HeadAndTail.createFrom(Flux.from(payloads).map(this::toMessage)))
-          .flatMap(
-              pair -> {
-                ServiceMessage message = pair.head();
-                validateRequest(message);
-                Flux<ServiceMessage> messages = Flux.from(pair.tail()).startWith(message);
-                ServiceMethodInvoker methodInvoker = methodRegistry.getInvoker(message.qualifier());
-                return methodInvoker.invokeBidirectional(messages, ServiceMessageCodec::decodeData);
+      return Flux.from(payloads)
+          .map(this::toMessage)
+          .switchOnFirst(
+              (first, messages) -> {
+                if (first.hasValue()) {
+                  ServiceMessage message = first.get();
+                  validateRequest(message);
+                  ServiceMethodInvoker methodInvoker =
+                      methodRegistry.getInvoker(message.qualifier());
+                  return methodInvoker.invokeBidirectional(
+                      messages, ServiceMessageCodec::decodeData);
+                }
+
+                return messages;
               })
           .map(this::toPayload);
     }

services/pom.xml

@@ -33,16 +33,6 @@
       <artifactId>jctools-core</artifactId>
     </dependency>
 
-    <dependency>
-      <groupId>org.hamcrest</groupId>
-      <artifactId>hamcrest-all</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>io.projectreactor</groupId>
-      <artifactId>reactor-test</artifactId>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>io.scalecube</groupId>
       <artifactId>scalecube-services-discovery</artifactId>

services/src/main/java/io/scalecube/services/ServiceCall.java

@@ -201,23 +201,27 @@ public Flux<ServiceMessage> requestBidirectional(Publisher<ServiceMessage> publi
    */
   public Flux<ServiceMessage> requestBidirectional(
       Publisher<ServiceMessage> publisher, Class<?> responseType) {
-    return Flux.from(HeadAndTail.createFrom(publisher))
-        .flatMap(
-            pair -> {
-              ServiceMessage request = pair.head();
-              String qualifier = request.qualifier();
-              Flux<ServiceMessage> messages = Flux.from(pair.tail()).startWith(request);
-
-              if (methodRegistry.containsInvoker(qualifier)) { // local service.
-                return methodRegistry
-                    .getInvoker(qualifier)
-                    .invokeBidirectional(messages, ServiceMessageCodec::decodeData)
-                    .map(this::throwIfError);
-              } else {
-                // remote service
-                return addressLookup(request)
-                    .flatMapMany(address -> requestBidirectional(messages, responseType, address));
+    return Flux.from(publisher)
+        .switchOnFirst(
+            (first, messages) -> {
+              if (first.hasValue()) {
+                ServiceMessage request = first.get();
+                String qualifier = request.qualifier();
+
+                if (methodRegistry.containsInvoker(qualifier)) { // local service.
+                  return methodRegistry
+                      .getInvoker(qualifier)
+                      .invokeBidirectional(messages, ServiceMessageCodec::decodeData)
+                      .map(this::throwIfError);
+                } else {
+                  // remote service
+                  return addressLookup(request)
+                      .flatMapMany(
+                          address -> requestBidirectional(messages, responseType, address));
+                }
               }
+
+              return messages;
             });
   }