package io.scalecube.gateway.websocket;

import io.scalecube.services.Microservices;
import io.scalecube.services.ServiceCall;
import io.scalecube.services.api.ServiceMessage;
import io.scalecube.services.codec.ServiceMessageDataCodec;
import io.scalecube.services.exceptions.ExceptionProcessor;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

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

import java.net.InetSocketAddress;
import java.time.Duration;

public class WebSocketServerEchoRunner {

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

  /**
   * Run test runner of Websocket server.
   * @param args - program arguments if any.
   * @throws InterruptedException - in case the program was interrupted.
   */
  public static void main(String[] args) throws InterruptedException {

    GreetingService serviceInstance = new GreetingServiceImpl();

    Microservices services = Microservices.builder()
        .services(serviceInstance).build().startAwait();
    ServiceCall.Call call = services.call();
    LOGGER.info("Started services at address: {}", services.serviceAddress());

    ServiceMessageDataCodec dataCodec = new ServiceMessageDataCodec();

    WebSocketAcceptor acceptor = new WebSocketAcceptor() {

      @Override
      public Mono<Void> onConnect(WebSocketSession session) {
        Flux<ServiceMessage> respStream = session
            .receive().log("###.receive()")
            .concatMap(call::requestMany).log("###.transform()")
            .onErrorResume(throwable -> Mono.just(ExceptionProcessor.toMessage(throwable)));
        
        return session.send(respStream
                .map(dataCodec::encode)
                .log())
            .then();
        
       
      }

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

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

}

@@ -1390,93 +1390,28 @@
     
-Flux%3CServiceMessage%3E respStream = session%0A            .receive().log(%22###.receive()%22)
+return session.send(
 %0A   
@@ -1423,203 +1423,107 @@
     
-.concatMap(
 call
-::
+.
 request
-Many).log(%22###.transform()%22)%0A            .onErrorResume(throwable -%3E Mono.just(ExceptionProcessor.toMessage(throwable)));%0A        %0A        return session.send(respStream%0A 
+Bidirectional(session.receive()%0A                .log(%22recv%22))%0A            .log(%22send%22)%0A
     
@@ -1530,19 +1530,16 @@
         
-   
 .map(dat
@@ -1557,70 +1557,17 @@
 ode)
-%0A                .log())%0A            .then();%0A        %0A       
+).then();
 %0A   

make it work bidirectional