<commit_before>package com.jetbrains.python.console.thrift.client

import com.jetbrains.python.console.thrift.DirectedMessage
import com.jetbrains.python.console.thrift.DirectedMessageCodec
import com.jetbrains.python.console.thrift.TCumulativeTransport
import io.netty.bootstrap.Bootstrap
import io.netty.buffer.ByteBuf
import io.netty.buffer.Unpooled
import io.netty.channel.Channel
import io.netty.channel.ChannelInitializer
import io.netty.channel.ChannelOption
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.NioSocketChannel
import io.netty.handler.codec.LengthFieldBasedFrameDecoder
import io.netty.handler.codec.LengthFieldPrepender
import io.netty.handler.logging.LogLevel
import io.netty.handler.logging.LoggingHandler
import org.apache.thrift.transport.TServerTransport
import org.apache.thrift.transport.TTransport
import java.io.PipedInputStream
import java.io.PipedOutputStream
import java.util.concurrent.atomic.AtomicBoolean

class TNettyClientTransport(private val host: String,
                            private val port: Int) : TTransport() {
  private var channel: Channel? = null

  private var messageHandler: DirectedMessageHandler? = null

  private val responseStream: PipedOutputStream = PipedOutputStream()

  private val responseInputStream: PipedInputStream = PipedInputStream(responseStream)

  /**
   * "Client as server" accepted transport.
   */
  private val serverAcceptedTransport = TNettyTransport()

  val serverTransport: TServerTransport = object : TServerTransport() {
    private val acceptedOnce = AtomicBoolean(false)

    override fun listen() {
      // TODO ?
    }

    override fun acceptImpl(): TTransport {
      if (acceptedOnce.compareAndSet(false, true)) {
        return serverAcceptedTransport
      }

      val lock = Object()
      while (true) {
        synchronized(lock) {
          lock.wait()
        }
      }
    }

    override fun close() {
      // TODO ?
    }
  }

  private fun assureChannelPresent(): Channel = channel ?: throw IllegalStateException("`channel` is `null`")

  private val messageBuffer: ByteBuf = Unpooled.buffer()

  override fun open() {
    // TODO check state

    val workerGroup = NioEventLoopGroup()

    try {
      val b = Bootstrap() // (1)
      b.group(workerGroup) // (2)
      b.channel(NioSocketChannel::class.java) // (3)
      b.option(ChannelOption.SO_KEEPALIVE, true) // (4)
      b.handler(object : ChannelInitializer<SocketChannel>() {
        override fun initChannel(ch: SocketChannel) {
          ch.pipeline().addLast(LoggingHandler(LogLevel.DEBUG))

          ch.pipeline().addLast(LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4))
          ch.pipeline().addLast(LengthFieldPrepender(4))

          ch.pipeline().addLast(DirectedMessageCodec())
          ch.pipeline().addLast(DirectedMessageHandler(responseStream, serverAcceptedTransport.outputStream))
        }
      })

      // Start the client.
      val f = b.connect(host, port).sync() // (5)

      channel = f.channel()

      messageHandler = f.channel().pipeline().get(DirectedMessageHandler::class.java)

      /*
            // Wait until the connection is closed.
            f.channel().closeFuture().sync()
      */
    }
    finally {
      /*
            workerGroup.shutdownGracefully()
      */
    }
  }

  override fun write(buf: ByteArray, off: Int, len: Int) {
    messageBuffer.writeBytes(buf, off, len)
  }

  /**
   * This is a nightmare but [flush] is used as the indicator of the message
   * write termination... by Thrift itself!
   */
  override fun flush() {
    try {
      messageHandler?.sendRequest(messageBuffer)
    }
    finally {
      messageBuffer.clear()
    }
  }

  override fun isOpen(): Boolean = channel?.isOpen == true

  override fun close() {
    // Wait until the connection is closed.
    assureChannelPresent().closeFuture().sync()
  }<commit_msg>PY-18029 Remove unused todos and improve class documentation<commit_after>package com.jetbrains.python.console.thrift.client

import com.jetbrains.python.console.thrift.DirectedMessage
import com.jetbrains.python.console.thrift.DirectedMessageCodec
import com.jetbrains.python.console.thrift.TCumulativeTransport
import io.netty.bootstrap.Bootstrap
import io.netty.buffer.ByteBuf
import io.netty.buffer.Unpooled
import io.netty.channel.Channel
import io.netty.channel.ChannelInitializer
import io.netty.channel.ChannelOption
import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.NioSocketChannel
import io.netty.handler.codec.LengthFieldBasedFrameDecoder
import io.netty.handler.codec.LengthFieldPrepender
import io.netty.handler.logging.LogLevel
import io.netty.handler.logging.LoggingHandler
import org.apache.thrift.transport.TServerTransport
import org.apache.thrift.transport.TTransport
import java.io.PipedInputStream
import java.io.PipedOutputStream
import java.util.concurrent.atomic.AtomicBoolean

class TNettyClientTransport(private val host: String,
                            private val port: Int) : TTransport() {
  private var channel: Channel? = null

  private var messageHandler: DirectedMessageHandler? = null

  private val responseStream: PipedOutputStream = PipedOutputStream()

  private val responseInputStream: PipedInputStream = PipedInputStream(responseStream)

  /**
   * "Client as server" accepted transport.
   */
  private val serverAcceptedTransport = TNettyTransport()

  val serverTransport: TServerTransport = object : TServerTransport() {
    private val acceptedOnce = AtomicBoolean(false)

    override fun listen() {}

    override fun acceptImpl(): TTransport {
      if (acceptedOnce.compareAndSet(false, true)) {
        return serverAcceptedTransport
      }

      // waits forever for the consequent `accept()` calls
      // in `org.apache.thrift.server.TThreadPoolServer.serve`
      val lock = Object()
      while (true) {
        synchronized(lock) {
          lock.wait()
        }
      }
    }

    override fun close() {}
  }

  private fun assureChannelPresent(): Channel = channel ?: throw IllegalStateException("`channel` is `null`")

  private val messageBuffer: ByteBuf = Unpooled.buffer()

  override fun open() {
    val workerGroup = NioEventLoopGroup()

    try {
      val b = Bootstrap() // (1)
      b.group(workerGroup) // (2)
      b.channel(NioSocketChannel::class.java) // (3)
      b.option(ChannelOption.SO_KEEPALIVE, true) // (4)
      b.handler(object : ChannelInitializer<SocketChannel>() {
        override fun initChannel(ch: SocketChannel) {
          ch.pipeline().addLast(LoggingHandler(LogLevel.DEBUG))

          ch.pipeline().addLast(LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4))
          ch.pipeline().addLast(LengthFieldPrepender(4))

          ch.pipeline().addLast(DirectedMessageCodec())
          ch.pipeline().addLast(DirectedMessageHandler(responseStream, serverAcceptedTransport.outputStream))
        }
      })

      // Start the client.
      val f = b.connect(host, port).sync() // (5)

      channel = f.channel()

      messageHandler = f.channel().pipeline().get(DirectedMessageHandler::class.java)

      /*
            // Wait until the connection is closed.
            f.channel().closeFuture().sync()
      */
    }
    finally {
      /*
            workerGroup.shutdownGracefully()
      */
    }
  }

  override fun write(buf: ByteArray, off: Int, len: Int) {
    messageBuffer.writeBytes(buf, off, len)
  }

  /**
   * Sends the request to the server with the buffer content.
   *
   * Note that the buffer might contain arbitrary portion of consecutive
   * bytes of the request message or messages.
   */
  override fun flush() {
    try {
      messageHandler?.sendRequest(messageBuffer)
    }
    finally {
      messageBuffer.clear()
    }
  }

  override fun isOpen(): Boolean = channel?.isOpen == true

  override fun close() {
    // Wait until the connection is closed.
    assureChannelPresent().closeFuture().sync()
  }
{'fraction_non_alphanumeric': 0.10458488228004957, 'fraction_numerical': 0.0048327137546468404, 'mean_word_length': 3.25910290237467, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '5400868', 'n_tokens_mistral': 2325, 'n_tokens_neox': 2276, 'n_words': 501}