<commit_before>package io.scalecube.services.transport.rsocket.client;

import io.scalecube.services.codec.ServiceMessageCodec;
import io.scalecube.services.transport.client.api.ClientChannel;
import io.scalecube.services.transport.client.api.ClientTransport;
import io.scalecube.transport.Address;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.client.TcpClientTransport;

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

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.ipc.netty.tcp.TcpClient;

public class RSocketClientTransport implements ClientTransport {

  private static final Logger LOGGER = LoggerFactory.getLogger(RSocketClientTransport.class);

  private final ThreadLocal<Map<Address, Mono<RSocket>>> rSockets = ThreadLocal.withInitial(ConcurrentHashMap::new);

  private final ServiceMessageCodec codec;

  public RSocketClientTransport(ServiceMessageCodec codec) {
    this.codec = codec;
  }

  @Override
  public ClientChannel create(Address address) {
    final Map<Address, Mono<RSocket>> monoMap = rSockets.get(); // keep reference for threadsafety
    Mono<RSocket> rSocket = monoMap.computeIfAbsent(address, address1 -> connect(address1, monoMap));
    return new RSocketServiceClientAdapter(rSocket, codec);
  }

  private static Mono<RSocket> connect(Address address, Map<Address, Mono<RSocket>> monoMap) {
    MonoProcessor<Scheduler> schedulerProcessor = MonoProcessor.create();
    MonoProcessor<RSocket> rSocketProcessor = MonoProcessor.create();

    RSocketFactory.connect()
        .transport(createTcpClientTransport(monoMap, address, schedulerProcessor))
        .start()
        .subscribe(
            rSocket -> {
              LOGGER.debug("Connected successfully on {}", address);
              rSocket.onClose().subscribe(aVoid -> {
                monoMap.remove(address);
                LOGGER.debug("Connection closed on {} and removed from the pool", address);
              });
              rSocketProcessor.onNext(rSocket);
              rSocketProcessor.onComplete();
            },
            throwable -> {
              LOGGER.warn("Connect failed on {}, cause: {}", address, throwable);
              monoMap.remove(address);
              rSocketProcessor.onError(throwable);
            });

    return schedulerProcessor.flatMap(rSocketProcessor::subscribeOn);
  }

  private static TcpClientTransport createTcpClientTransport(Map<Address, Mono<RSocket>> monoMap,
      Address address, MonoProcessor<Scheduler> schedulerProcessor) {

    return TcpClientTransport.create(
        TcpClient.create(options -> options
            .disablePool()
            .host(address.host())
            .port(address.port())
            .afterNettyContextInit(nettyContext -> {
              // add handler to react on remote node closes connection
              nettyContext.addHandler(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelInactive(ChannelHandlerContext ctx) {
                  monoMap.remove(address);
                  LOGGER.debug("Connection became inactive on {} and removed from monoMap", address);
                  ctx.fireChannelInactive();
                }
              });
              // expose executor where channel was assigned
              Scheduler scheduler = Schedulers.fromExecutor(nettyContext.channel().eventLoop());
              schedulerProcessor.onNext(scheduler);
              schedulerProcessor.onComplete();
              LOGGER.debug("Obtained scheduler {} on channel {}", scheduler, nettyContext.channel());
            })));
  }
}
<commit_msg>Removed ThreadLocal from RSocketClientTransport.java; were causing problems at RequestManyBenchmarks<commit_after>package io.scalecube.services.transport.rsocket.client;

import io.scalecube.services.codec.ServiceMessageCodec;
import io.scalecube.services.transport.client.api.ClientChannel;
import io.scalecube.services.transport.client.api.ClientTransport;
import io.scalecube.transport.Address;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.client.TcpClientTransport;

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

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.ipc.netty.tcp.TcpClient;

public class RSocketClientTransport implements ClientTransport {

  private static final Logger LOGGER = LoggerFactory.getLogger(RSocketClientTransport.class);

  private final Map<Address, Mono<RSocket>> rSockets = new ConcurrentHashMap<>();

  private final ServiceMessageCodec codec;

  public RSocketClientTransport(ServiceMessageCodec codec) {
    this.codec = codec;
  }

  @Override
  public ClientChannel create(Address address) {
    final Map<Address, Mono<RSocket>> monoMap = rSockets; // keep reference for threadsafety
    Mono<RSocket> rSocket = monoMap.computeIfAbsent(address, address1 -> connect(address1, monoMap));
    return new RSocketServiceClientAdapter(rSocket, codec);
  }

  private static Mono<RSocket> connect(Address address, Map<Address, Mono<RSocket>> monoMap) {
    MonoProcessor<Scheduler> schedulerProcessor = MonoProcessor.create();
    MonoProcessor<RSocket> rSocketProcessor = MonoProcessor.create();

    RSocketFactory.connect()
        .transport(createTcpClientTransport(monoMap, address, schedulerProcessor))
        .start()
        .subscribe(
            rSocket -> {
              LOGGER.debug("Connected successfully on {}", address);
              rSocket.onClose().subscribe(aVoid -> {
                monoMap.remove(address);
                LOGGER.debug("Connection closed on {} and removed from the pool", address);
              });
              rSocketProcessor.onNext(rSocket);
              rSocketProcessor.onComplete();
            },
            throwable -> {
              LOGGER.warn("Connect failed on {}, cause: {}", address, throwable);
              monoMap.remove(address);
              rSocketProcessor.onError(throwable);
            });

    return schedulerProcessor.flatMap(rSocketProcessor::subscribeOn);
  }

  private static TcpClientTransport createTcpClientTransport(Map<Address, Mono<RSocket>> monoMap,
      Address address, MonoProcessor<Scheduler> schedulerProcessor) {

    return TcpClientTransport.create(
        TcpClient.create(options -> options
            .disablePool()
            .host(address.host())
            .port(address.port())
            .afterNettyContextInit(nettyContext -> {
              // add handler to react on remote node closes connection
              nettyContext.addHandler(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelInactive(ChannelHandlerContext ctx) {
                  monoMap.remove(address);
                  LOGGER.debug("Connection became inactive on {} and removed from monoMap", address);
                  ctx.fireChannelInactive();
                }
              });
              // expose executor where channel was assigned
              Scheduler scheduler = Schedulers.fromExecutor(nettyContext.channel().eventLoop());
              schedulerProcessor.onNext(scheduler);
              schedulerProcessor.onComplete();
              LOGGER.debug("Obtained scheduler {} on channel {}", scheduler, nettyContext.channel());
            })));
  }
}

{'fraction_non_alphanumeric': 0.09596977329974811, 'fraction_numerical': 0.0010075566750629723, 'mean_word_length': 3.1905013192612137, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3146803', 'n_tokens_mistral': 2091, 'n_tokens_neox': 2054, 'n_words': 407}