package io.scalecube.services.transport.rsocket.server;

import io.scalecube.services.transport.rsocket.PayloadCodec;
import io.scalecube.services.transport.server.api.ServerMessageAcceptor;

import io.rsocket.AbstractRSocket;
import io.rsocket.ConnectionSetupPayload;
import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.SocketAcceptor;

import org.reactivestreams.Publisher;

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

public class RSocketServiceMethodAcceptor implements SocketAcceptor {

  private PayloadCodec payloadCodec;
  private ServerMessageAcceptor acceptor;
 
  public RSocketServiceMethodAcceptor(ServerMessageAcceptor acceptor, PayloadCodec payloadCodec) {
   this.payloadCodec = payloadCodec;
   this.acceptor = acceptor;
  }

  @Override
  public Mono<RSocket> accept(ConnectionSetupPayload setup, RSocket sendingSocket) {
    
    return Mono.just(new AbstractRSocket() {
      @Override
      public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
        return null;  
      }

      @Override
      public Flux<Payload> requestStream(Payload payload) {
        return null;
      }

      @Override
      public Mono<Payload> requestResponse(Payload payload) {

        return null;
      }

      @Override
      public Mono<Void> fireAndForget(Payload payload) {
        return null;
      }
    });
  }
}

@@ -1136,36 +1136,142 @@
 %0A        return 
-null
+acceptor.requestStream(payloadCodec.decode(payload))%0A            .map(response-%3EpayloadCodec.encode(response))
 ;%0A      %7D%0A%0A     
@@ -1343,17 +1343,16 @@
 load) %7B%0A
-%0A
         
@@ -1350,36 +1350,144 @@
 %0A        return 
-null
+acceptor.requestResponse(payloadCodec.decode(payload))%0A            .map(response-%3EpayloadCodec.encode(response))
 ;%0A      %7D%0A%0A     
@@ -1569,20 +1569,68 @@
  return 
-null
+acceptor.fireAndForget(payloadCodec.decode(payload))
 ;%0A      

use codec in socket acceptor