package io.scalecube.gateway.websocket;

import io.scalecube.services.Microservices;
import io.scalecube.services.ServiceCall;
import io.scalecube.services.annotations.Service;
import io.scalecube.services.annotations.ServiceMethod;
import io.scalecube.services.api.ServiceMessage;
import io.scalecube.services.codec.ServiceMessageDataCodec;
import io.scalecube.services.exceptions.ExceptionProcessor;
import io.scalecube.transport.Address;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;

public class WebSocketServerEchoRunner {

  public static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServerEchoRunner.class);

  @Service("greeting")
  public interface ServiceHello {
    @ServiceMethod
    Mono<String> helloString(String hello);
  }

  public static class ServiceHelloImpl implements ServiceHello {
    @Override
    public Mono<String> helloString(String hello) {
      return Mono.just(hello).log("^^^^^^^^^ ServiceHelloImpl");
    }
  }

  public static void main(String[] args) throws InterruptedException {

    Microservices gateway = Microservices.builder().build().startAwait();
    Address address = gateway.cluster().address();

    LOGGER.info("address: " + address);

    Microservices services = Microservices.builder()
        .seeds(address)
        .services(new ServiceHelloImpl())
        .build()
        .startAwait();

    LOGGER.info("services: " + services);

    ServiceMessageDataCodec dataCodec = new ServiceMessageDataCodec();

    WebSocketAcceptor acceptor = new WebSocketAcceptor() {
      @Override
      public Mono<Void> onConnect(WebSocketSession session) {
        ServiceCall.Call call = services.call();

        MonoProcessor<ServiceMessage> requestProcessor = MonoProcessor.create();
        MonoProcessor<ServiceMessage> responseProcessor = MonoProcessor.create();

        session.receive().subscribe(requestProcessor::onNext);

        call.requestBidirectional(requestProcessor)
            .onErrorResume(throwable -> Mono.just(ExceptionProcessor.toMessage(throwable)))
            .map(dataCodec::encode)
            .subscribe(responseProcessor::onNext);

        return session.send(responseProcessor);
      }

      @Override
      public Mono<Void> onDisconnect(WebSocketSession session) {
        return Mono.never();
      }
    };

    WebSocketServer server = new WebSocketServer(acceptor);
    server.start(new InetSocketAddress(8080));
    Thread.currentThread().join();
  }
}

@@ -525,24 +525,153 @@
 etAddress;%0A%0A
+import reactor.core.publisher.EmitterProcessor;%0Aimport reactor.core.publisher.Flux;%0Aimport reactor.core.publisher.FluxProcessor;%0A
 import react
@@ -1131,24 +1131,96 @@
 ng hello) %7B%0A
+      System.out.println(%22ServiceHelloImpl/helloString say: %22 + hello);%0A
       return
@@ -1926,32 +1926,33 @@
 sion session) %7B%0A
+%0A
         ServiceC
@@ -1984,25 +1984,24 @@
 call();%0A
-%0A
         
 MonoProc
@@ -1984,36 +1984,36 @@
 call();%0A        
-Mono
+Flux
 Processor%3CServic
@@ -1998,32 +1998,48 @@
   FluxProcessor%3C
+ServiceMessage, 
 ServiceMessage%3E 
@@ -2049,36 +2049,39 @@
 uestProcessor = 
-Mono
+Emitter
 Processor.create
@@ -2092,20 +2092,20 @@
         
-Mono
+Flux
 Processo
@@ -2098,32 +2098,48 @@
   FluxProcessor%3C
+ServiceMessage, 
 ServiceMessage%3E 
@@ -2158,20 +2158,23 @@
 essor = 
-Mono
+Emitter
 Processo
@@ -2181,25 +2181,24 @@
 r.create();%0A
-%0A
         sess
@@ -2261,52 +2261,79 @@
     
-call.
 request
-Bidirectional(requestProcessor)%0A
+Processor.subscribe(next -%3E %7B%0A          call.requestOne(next)%0A  
     
@@ -2432,16 +2432,18 @@
         
+  
 .map(dat
@@ -2458,16 +2458,18 @@
 encode)%0A
+  
         
@@ -2503,32 +2503,43 @@
 essor::onNext);%0A
+        %7D);
 %0A        return 

fix the example