Merge pull request #148 from scalecube/fix-cancel-signal-on-client

Fix sending cancel signal on client

Author: segabriel

services-gateway-client-transport/src/main/java/io/scalecube/services/gateway/transport/websocket/WebsocketGatewayClient.java

@@ -90,7 +90,7 @@ public Mono<ServiceMessage> requestResponse(ServiceMessage request) {
                           .send(encodeRequest(request, sid), sid)
                           .doOnSubscribe(s -> LOGGER.debug("Sending request {}", request))
                           .then(session.newMonoProcessor(sid))
-                          .doOnCancel(() -> handleCancel(sid, session))
+                          .doOnCancel(() -> handleCancel(sid, request.qualifier(), session))
                           .doFinally(s -> session.removeProcessor(sid)));
         });
   }
@@ -107,7 +107,7 @@ public Flux<ServiceMessage> requestStream(ServiceMessage request) {
                           .send(encodeRequest(request, sid), sid)
                           .doOnSubscribe(s -> LOGGER.debug("Sending request {}", request))
                           .thenMany(session.newUnicastProcessor(sid))
-                          .doOnCancel(() -> handleCancel(sid, session))
+                          .doOnCancel(() -> handleCancel(sid, request.qualifier(), session))
                           .doFinally(s -> session.removeProcessor(sid)));
         });
   }
@@ -211,10 +211,11 @@ private void onReadIdle(Connection connection) {
         .subscribe(null, ex -> LOGGER.warn("Can't send keepalive on readIdle: " + ex));
   }
 
-  private void handleCancel(long sid, WebsocketGatewayClientSession session) {
+  private void handleCancel(long sid, String qualifier, WebsocketGatewayClientSession session) {
     ByteBuf byteBuf =
         codec.encode(
             ServiceMessage.builder()
+                .qualifier(qualifier)
                 .header(STREAM_ID, sid)
                 .header(SIGNAL, Signal.CANCEL.codeAsString())
                 .build());