<commit_before>package org.jetbrains.bsp.protocol

import java.io.{InputStream, OutputStream}
import java.util.concurrent.{CompletableFuture, LinkedBlockingQueue, TimeUnit, TimeoutException}

import ch.epfl.scala.bsp4j
import com.intellij.notification.NotificationType
import com.intellij.openapi.diagnostic.Logger
import com.intellij.util.concurrency.AppExecutorUtil
import org.eclipse.lsp4j.jsonrpc.{Launcher, ResponseErrorException}
import org.jetbrains.bsp._
import org.jetbrains.bsp.protocol.BspNotifications._
import org.jetbrains.bsp.protocol.BspSession._

import scala.collection.mutable.ListBuffer
import scala.concurrent._
import scala.concurrent.duration._
import scala.util.{Failure, Success, Try}
import scala.util.control.NonFatal

class BspSession(bspIn: InputStream,
                 bspOut: OutputStream,
                 initializeBuildParams: bsp4j.InitializeBuildParams,
                 cleanup: ()=>Unit
                ) {

  private val logger = Logger.getInstance(classOf[BspCommunication])

  private val jobs = new LinkedBlockingQueue[BspSessionJob[_,_]]

  private var currentJob: BspSessionJob[_,_] = DummyJob
  private var notificationCallbacks: List[NotificationCallback] = Nil

  private val serverConnection: ServerConnection = startServerConnection
  private val sessionInitialized = initializeSession
  private val sessionShutdown = Promise[Unit]

  private val queuePause = 10.millis
  private val queueTimeout = 1.second
  private val sessionTimeout = 5.seconds

  private val queueProcessor = AppExecutorUtil.getAppScheduledExecutorService
      .scheduleWithFixedDelay(() => nextQueuedCommand, queuePause.toMillis, queuePause.toMillis, TimeUnit.MILLISECONDS)

  private def notifications(notification: BspNotification): Unit =
    notificationCallbacks.foreach(_.apply(notification))

  private def nextQueuedCommand= {
    try {
      waitForSession(sessionTimeout)
    } catch {
      case to : TimeoutException =>
        val error = BspConnectionError("bsp server is not responding", to)
        logger.warn(error)
        shutdown(Some(error))
      case NonFatal(error) =>
        val msg = "problem connecting to bsp server"
        val bspError = BspException(msg, error)
        logger.warn(bspError)
        shutdown(Some(bspError))
    }

    try {
      val currentIgnoringErrors = currentJob.future.recover {
        case NonFatal(_) => ()
      }(ExecutionContext.global)
      Await.result(currentIgnoringErrors, queueTimeout) // will throw on job error

      val next = jobs.poll(queueTimeout.toMillis, TimeUnit.MILLISECONDS)
      if (next != null) {
        currentJob = next
        currentJob.run(serverConnection.server)
      }
    } catch {
      case _: TimeoutException => // just carry on
      case NonFatal(error) =>
        val bspError = BspException("problem executing bsp job", error)
        logger.error(bspError)
        currentJob.cancelWithError(bspError)
    }
  }

  private def startServerConnection: ServerConnection = {

    val localClient = new BspSessionClient

    val launcher = new Launcher.Builder[BspServer]()
      .setRemoteInterface(classOf[BspServer])
      .setExecutorService(AppExecutorUtil.getAppExecutorService)
      .setInput(bspIn)
      .setOutput(bspOut)
      .setLocalService(localClient)
      .create()
    val listening = launcher.startListening()
    val bspServer = launcher.getRemoteProxy
    localClient.onConnectWithServer(bspServer)

    val cancelable = Cancelable { () =>
      Cancelable.cancelAll(
        List(
          Cancelable(() => bspIn.close()),
          Cancelable(() => bspOut.close()),
          Cancelable(() => listening.cancel(true)),
          Cancelable(() => cleanup())
        )
      )
    }

    ServerConnection(bspServer, cancelable)
  }

  private def initializeSession: CompletableFuture[bsp4j.InitializeBuildResult] = {
    val bspServer = serverConnection.server
    bspServer.buildInitialize(initializeBuildParams)
      .thenApply[bsp4j.InitializeBuildResult] { result =>
      bspServer.onBuildInitialized()
        result
      }
      .exceptionally {
        case responseError: ResponseErrorException => throw BspConnectionError(responseError.getMessage, responseError)
      }<commit_msg>wrap all exceptions on bsp session init #SCL-14553<commit_after>package org.jetbrains.bsp.protocol

import java.io.{InputStream, OutputStream}
import java.util.concurrent.{CompletableFuture, LinkedBlockingQueue, TimeUnit, TimeoutException}

import ch.epfl.scala.bsp4j
import com.intellij.notification.NotificationType
import com.intellij.openapi.diagnostic.Logger
import com.intellij.util.concurrency.AppExecutorUtil
import org.eclipse.lsp4j.jsonrpc.{Launcher, ResponseErrorException}
import org.jetbrains.bsp._
import org.jetbrains.bsp.protocol.BspNotifications._
import org.jetbrains.bsp.protocol.BspSession._

import scala.collection.mutable.ListBuffer
import scala.concurrent._
import scala.concurrent.duration._
import scala.util.{Failure, Success, Try}
import scala.util.control.NonFatal

class BspSession(bspIn: InputStream,
                 bspOut: OutputStream,
                 initializeBuildParams: bsp4j.InitializeBuildParams,
                 cleanup: ()=>Unit
                ) {

  private val logger = Logger.getInstance(classOf[BspCommunication])

  private val jobs = new LinkedBlockingQueue[BspSessionJob[_,_]]

  private var currentJob: BspSessionJob[_,_] = DummyJob
  private var notificationCallbacks: List[NotificationCallback] = Nil

  private val serverConnection: ServerConnection = startServerConnection
  private val sessionInitialized = initializeSession
  private val sessionShutdown = Promise[Unit]

  private val queuePause = 10.millis
  private val queueTimeout = 1.second
  private val sessionTimeout = 5.seconds

  private val queueProcessor = AppExecutorUtil.getAppScheduledExecutorService
      .scheduleWithFixedDelay(() => nextQueuedCommand, queuePause.toMillis, queuePause.toMillis, TimeUnit.MILLISECONDS)

  private def notifications(notification: BspNotification): Unit =
    notificationCallbacks.foreach(_.apply(notification))

  private def nextQueuedCommand= {
    try {
      waitForSession(sessionTimeout)
    } catch {
      case to : TimeoutException =>
        val error = BspConnectionError("bsp server is not responding", to)
        logger.warn(error)
        shutdown(Some(error))
      case NonFatal(error) =>
        val msg = "problem connecting to bsp server"
        val bspError = BspException(msg, error)
        logger.warn(bspError)
        shutdown(Some(bspError))
    }

    try {
      val currentIgnoringErrors = currentJob.future.recover {
        case NonFatal(_) => ()
      }(ExecutionContext.global)
      Await.result(currentIgnoringErrors, queueTimeout) // will throw on job error

      val next = jobs.poll(queueTimeout.toMillis, TimeUnit.MILLISECONDS)
      if (next != null) {
        currentJob = next
        currentJob.run(serverConnection.server)
      }
    } catch {
      case _: TimeoutException => // just carry on
      case NonFatal(error) =>
        val bspError = BspException("problem executing bsp job", error)
        logger.error(bspError)
        currentJob.cancelWithError(bspError)
    }
  }

  private def startServerConnection: ServerConnection = {

    val localClient = new BspSessionClient

    val launcher = new Launcher.Builder[BspServer]()
      .setRemoteInterface(classOf[BspServer])
      .setExecutorService(AppExecutorUtil.getAppExecutorService)
      .setInput(bspIn)
      .setOutput(bspOut)
      .setLocalService(localClient)
      .create()
    val listening = launcher.startListening()
    val bspServer = launcher.getRemoteProxy
    localClient.onConnectWithServer(bspServer)

    val cancelable = Cancelable { () =>
      Cancelable.cancelAll(
        List(
          Cancelable(() => bspIn.close()),
          Cancelable(() => bspOut.close()),
          Cancelable(() => listening.cancel(true)),
          Cancelable(() => cleanup())
        )
      )
    }

    ServerConnection(bspServer, cancelable)
  }

  private def initializeSession: CompletableFuture[bsp4j.InitializeBuildResult] = {
    val bspServer = serverConnection.server
    bspServer.buildInitialize(initializeBuildParams)
      .thenApply[bsp4j.InitializeBuildResult] { result =>
      bspServer.onBuildInitialized()
        result
      }
      .exceptionally {
        case NonFatal(error) => throw BspConnectionError(error.getMessage, error)
      }
{'fraction_non_alphanumeric': 0.08898055391868002, 'fraction_numerical': 0.0027106658809664115, 'mean_word_length': 3.533119658119658, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 31, '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': '7243759', 'n_tokens_mistral': 2441, 'n_tokens_neox': 2379, 'n_words': 488}