package io.scalecube.services.transport.rsocket;

import io.scalecube.services.transport.api.ClientTransport;
import io.scalecube.services.transport.api.HeadersCodec;
import io.scalecube.services.transport.api.ServerTransport;
import io.scalecube.services.transport.api.ServiceMessageCodec;
import io.scalecube.services.transport.api.ServiceTransport;
import io.scalecube.services.transport.api.TransportResources;
import reactor.netty.resources.LoopResources;

/**
 * RSocket service transport. Entry point for getting {@link RSocketClientTransport} and {@link
 * RSocketServerTransport}.
 */
public class RSocketServiceTransport implements ServiceTransport {

  public static final RSocketServiceTransport INSTANCE = new RSocketServiceTransport();

  private final ServiceMessageCodec messageCodec;
  private final LoopResources loopResources;

  public RSocketServiceTransport() {
    HeadersCodec headersCodec = HeadersCodec.getInstance("application/json");
    messageCodec =  new ServiceMessageCodec(headersCodec);
    loopResources =  LoopResources.create("rsocket-worker");
  }

  public RSocketServiceTransport(HeadersCodec headersCodec, LoopResources loopResources) {
    this.messageCodec = new ServiceMessageCodec(headersCodec);
    this.loopResources = loopResources;
  }

  @Override
  public ClientTransport clientTransport(TransportResources resources) {
    return new RSocketClientTransport(
            messageCodec,
        ((RSocketTransportResources) resources)
            .workerPool()
            .<LoopResources>map(DelegatedLoopResources::newClientLoopResources)
            .orElse(loopResources));
  }

  @Override
  public ServerTransport serverTransport(TransportResources resources) {
    return new RSocketServerTransport(
            messageCodec,
        ((RSocketTransportResources) resources)
            .workerPool()
            .<LoopResources>map(DelegatedLoopResources::newServerLoopResources)
            .orElse(loopResources));
  }
}

@@ -655,16 +655,56 @@
 port %7B%0A%0A
+  /**%0A   * Default Instance.%0A   *%0A   */%0A
   public
@@ -880,16 +880,56 @@
 urces;%0A%0A
+  /**%0A   * Default instance.%0A   *%0A   */%0A
   public
@@ -1053,17 +1053,16 @@
 eCodec =
- 
  new Ser
@@ -1112,17 +1112,16 @@
 ources =
- 
  LoopRes
@@ -1158,16 +1158,146 @@
 );%0A  %7D%0A%0A
+  /**%0A   * Constructor with DI.%0A   *%0A   * @param headersCodec  user's headers codec%0A   * @param loopResources loopResources%0A   */%0A
   public
@@ -1483,24 +1483,164 @@
 urces;%0A  %7D%0A%0A
+  /**%0A   * Fabric method for client transport.%0A   *%0A   * @param resources service transport resources%0A   * @return client's transport%0A   */%0A
   @Override%0A
@@ -1777,32 +1777,36 @@
 eCodec,%0A        
+    
 ((RSocketTranspo
@@ -1833,32 +1833,40 @@
 es)%0A            
+        
 .workerPool()%0A  
@@ -1867,32 +1867,40 @@
 l()%0A            
+ 
+       
 .%3CLoopResources%3E
@@ -1955,32 +1955,40 @@
 es)%0A            
+     
+   
 .orElse(loopReso
@@ -2001,16 +2001,156 @@
 );%0A  %7D%0A%0A
+  /**%0A   * Fabric method for server transport.%0A   *%0A   * @param resources service transport resources%0A   * @return server's transport%0A   */%0A
   @Overr
@@ -2283,32 +2283,36 @@
   messageCodec,%0A
+    
         ((RSocke
@@ -2347,32 +2347,40 @@
 es)%0A            
+        
 .workerPool()%0A  
@@ -2369,32 +2369,40 @@
   .workerPool()%0A
+        
             .%3CLo
@@ -2457,32 +2457,40 @@
 rLoopResources)%0A
+        
             .orE

fix codestyle