<commit_before>package nvim

import java.io.InputStream
import java.io.OutputStream
import java.net.Socket
import java.util.concurrent.atomic.AtomicInteger

import scala.concurrent.Future
import scala.concurrent.Promise
import scala.util.Failure
import scala.util.Success
import scala.util.Try

import org.msgpack.core.MessagePack
import org.msgpack.core.MessageUnpacker

import com.typesafe.scalalogging.LazyLogging

import msgpack4z._
import nvim.internal._

/**
 * Manages the connection to Neovim. Communication is done through the
 * msgpack-rpc protocol, whose spec is here:
 *
 * [[https://github.com/msgpack-rpc/msgpack-rpc/blob/master/spec.md]]
 */
final class Connection(host: String, port: Int) extends LazyLogging {

  val ResponseId = 1
  val NotificationId = 2

  private var requests = Map[Int, Response ⇒ Unit]()
  private var notificationHandlers = List[Notification ⇒ Unit]()

  private val gen = new IdGenerator
  private val socket = new Socket(host, port)
  private val thread = new Thread(new Runnable {
    override def run() = {
      val unp = MessagePack.DEFAULT.newUnpacker(socket.getInputStream)
      val unpacker = new Msgpack07Unpacker(unp)
      readResp(unp, unpacker)

      logger.warn(s"Connection to $host:$port lost")
    }
  })
  thread.start()

  private def readResp(unp: MessageUnpacker, unpacker: Msgpack07Unpacker): Unit = {
    /*
     * The stupid API doesn't allow to do a look ahead, therefore we have to
     * use reflection to correct the internal position in the buffer.
     */
    def lookAheadType(): Int = {
      unpacker.unpackArrayHeader()
      val tpe = unpacker.unpackInt()
      val f = classOf[MessageUnpacker].getDeclaredField("position")
      f.setAccessible(true)
      val pos = f.getInt(unp)
      f.setInt(unp, pos-2)
      tpe
    }

    lookAheadType() match {
      case ResponseId ⇒
        MsgpackCodec[Response].unpack(unpacker) match {
          case scalaz.-\/(e) ⇒
            logger.error("Couldn't unpack response", e)

          case scalaz.\/-(resp) ⇒
            logger.debug(s"retrieved: $resp")
            requests.get(resp.id) match {
              case Some(f) ⇒
                f(resp)
              case None ⇒
                logger.warn(s"The following response is ignored because its ID '${resp.id}' is unexpected: $resp")
            }
        }

      case NotificationId ⇒
        MsgpackCodec[Notification].unpack(unpacker) match {
          case scalaz.-\/(e) ⇒
            logger.error("Couldn't unpack response", e)

          case scalaz.\/-(notification) ⇒<commit_msg>Fix threading bug due to concurrent access of a Map<commit_after>package nvim

import java.net.Socket
import java.util.concurrent.atomic.AtomicInteger

import scala.collection.concurrent.TrieMap
import scala.concurrent.Future
import scala.concurrent.Promise
import scala.util.Failure
import scala.util.Success
import scala.util.Try

import org.msgpack.core.MessagePack
import org.msgpack.core.MessageUnpacker

import com.typesafe.scalalogging.LazyLogging

import msgpack4z._
import nvim.internal._

/**
 * Manages the connection to Neovim. Communication is done through the
 * msgpack-rpc protocol, whose spec is here:
 *
 * [[https://github.com/msgpack-rpc/msgpack-rpc/blob/master/spec.md]]
 */
final class Connection(host: String, port: Int) extends LazyLogging {

  val ResponseId = 1
  val NotificationId = 2

  private val requests = TrieMap[Int, Response ⇒ Unit]()
  private var notificationHandlers = List[Notification ⇒ Unit]()

  private val gen = new IdGenerator
  private val socket = new Socket(host, port)
  private val thread = new Thread(new Runnable {
    override def run() = {
      val unp = MessagePack.DEFAULT.newUnpacker(socket.getInputStream)
      val unpacker = new Msgpack07Unpacker(unp)
      readResp(unp, unpacker)

      logger.warn(s"Connection to $host:$port lost")
    }
  })
  thread.start()

  private def readResp(unp: MessageUnpacker, unpacker: Msgpack07Unpacker): Unit = {
    /*
     * The stupid API doesn't allow to do a look ahead, therefore we have to
     * use reflection to correct the internal position in the buffer.
     */
    def lookAheadType(): Int = {
      unpacker.unpackArrayHeader()
      val tpe = unpacker.unpackInt()
      val f = classOf[MessageUnpacker].getDeclaredField("position")
      f.setAccessible(true)
      val pos = f.getInt(unp)
      f.setInt(unp, pos-2)
      tpe
    }

    lookAheadType() match {
      case ResponseId ⇒
        MsgpackCodec[Response].unpack(unpacker) match {
          case scalaz.-\/(e) ⇒
            logger.error("Couldn't unpack response", e)

          case scalaz.\/-(resp) ⇒
            logger.debug(s"retrieved: $resp")
            requests.remove(resp.id) match {
              case Some(f) ⇒
                f(resp)
              case None ⇒
                logger.warn(s"The following response is ignored because its ID '${resp.id}' is unexpected: $resp")
            }
        }

      case NotificationId ⇒
        MsgpackCodec[Notification].unpack(unpacker) match {
          case scalaz.-\/(e) ⇒
            logger.error("Couldn't unpack response", e)

          case scalaz.\/-(notification) ⇒
{'fraction_non_alphanumeric': 0.10218273131157041, 'fraction_numerical': 0.0030905930075333203, 'mean_word_length': 2.804555473916238, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, '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': '5312899', 'n_tokens_mistral': 1630, 'n_tokens_neox': 1584, 'n_words': 428}