alternative stdio handling

InteroProcess.scala

package intellij.haskell.external.repl

import java.io._
import java.util.concurrent.{ConcurrentLinkedDeque, SynchronousQueue}

import com.intellij.util.EnvironmentUtil

import scala.collection.JavaConverters._
import scala.collection.mutable.{ArrayBuffer, ListBuffer}
import scala.io._
import scala.sys.process._


object Test {
  def foo(i: Integer, response: ArrayBuffer[String]): Unit = {
    response.foreach(println)
  }
}

/*
sbt -jvm-debug 5005 -Dscala.color

import intellij.haskell.external.repl._
val ip = new InteroProcess()
ip.startIntero()

ip.exec(":?", Test.foo)
ip.exec(":show", Test.foo)
ip.exec(":show paths", Test.foo)
*/

class InteroProcess(val extraStartOptions: Seq[String] = Seq()) {

  private final val LineSeparator = '\n'

  private[this] var available = false

  private final val EndOfOutputIndicator = "^IntellijHaskell^"

  private[this] var interoInputStream: OutputStream = _

  private[this] var procOut: InputStream = _
  private[this] var procErr: InputStream = _

  // TODO optimization: some commands only need to get parsed for "Ok/not Ok"... no need for ArrayBuffer[String] !
  private[this] val responses = new ConcurrentLinkedDeque[ArrayBuffer[String]]
  private[this] val callbacks = new ConcurrentLinkedDeque[ArrayBuffer[String] => Unit]

  private[this] val syncResponse = new SynchronousQueue[ArrayBuffer[String]](true)

  def testAsync(i: Integer) = {
    this.isSync = false
    for (n <- 1 to i) {
      println("async command " + n)
      this.execAsync(":show paths", Test.foo(n, _))
    }
  }

  def testSync(i: Integer) = {
    this.isSync = true
    for (n <- 1 to i) {
      println("sync command " + n)
      val strings = this.execSync(":show paths")
      strings.foreach((s) => {
        println(s)
      })
    }
  }

  var isSync = false

  def startIntero(): Unit = {

    if (available) {
      println("Stack repl can not be started because it's already running or busy with starting")
      return
    }

    try {
      val command = (Seq("stack", "repl", "--with-ghc", "intero", "--verbosity", "warn", "--no-build", "--terminal", "--no-load") ++ extraStartOptions).mkString(" ")
      println(s"Stack repl will be started with command: $command")

      val process = Option(EnvironmentUtil.getEnvironmentMap) match {
        case None => Process(command, new File("."))
        case Some(envMap) => Process(command, new File("."), envMap.asScala.toArray: _*)
      }

      process.run(
        new ProcessIO(

          in => interoInputStream = in,

          (out: InputStream) => {
            procOut = out
            Source.fromInputStream(out).getLines.foreach(line => {
              if (isSync) dealWithOutSync(line) else dealWithOut(line)
            })
          },

          (err: InputStream) => {
            procErr = err
            Source.fromInputStream(err).getLines.foreach(s => {
              // TODO
            })
          }
        ))

      // TODO ? We have to wait...
      Thread.sleep(1000)

      //      writeOutputToLogInfo()

      writeToIntero(s""":set prompt "$EndOfOutputIndicator\\n"""")
      writeToIntero(":set -Wall")
      writeToIntero(":set -fdefer-typed-holes")

      available = true

      Thread.sleep(5000)

      println("Stack repl is started.")
    }
    catch {
      case e: Exception =>
        println("Could not start Stack repl. Make sure you have set right path to Stack in settings.")
        println(s"Error message while trying to start Stack repl: ${e.getMessage}")
        exit(forceExit = true)
    }
  }

  private def writeToIntero(command: String) = {
    interoInputStream.write(command.getBytes)
    interoInputStream.write(LineSeparator)
    interoInputStream.flush()
  }

  private def reachedEnd(s: String): Boolean = {
    s != null && s.startsWith(EndOfOutputIndicator)
  }

  private def dealWithOut(s: String) = {
    if (!responses.isEmpty) {
      val response = responses.peek()
      if (response.isEmpty || !reachedEnd(s)) {
        response.+=(s)
      } else {
        responses.remove()
        // TODO removePrompt + convertOutputToOneMessagePerLine
        callbacks.remove()(response)
      }
    }
  }

  var responseObj: ArrayBuffer[String] = null

  private def dealWithOutSync(s: String) = {
    if (responseObj == null) {
      responseObj = new ArrayBuffer[String]()
    }
    if (responseObj.isEmpty || !reachedEnd(s)) {
      responseObj.+=(s)
    } else {
      // TODO removePrompt + convertOutputToOneMessagePerLine
      syncResponse.put(responseObj)
      responseObj = null
    }
  }

  def execAsync(command: String, f: ArrayBuffer[String] => Unit): Unit = {
    if (!available) {
      println(s"Stack repl is not yet available. Command was: $command")
      return
    }

    try {
      this.synchronized {
        responses.add(new ArrayBuffer[String])
        callbacks.add(f)
        writeToIntero(command)
      }
    }
    catch {
      case e: Exception =>
        println(s"Error in communication with Stack repl: ${e.getMessage}. Check if your Haskell/Stack environment is working okay. Command was: $command")
        exit()
        None
    }
  }

  def execSync(command: String): ArrayBuffer[String] = {
    if (!available) {
      println(s"Stack repl is not yet available. Command was: $command")
      return null
    }

    try {
      this.synchronized {
        writeToIntero(command)
        return syncResponse.take()
      }
    }
    catch {
      case e: Exception =>
        println(s"Error in communication with Stack repl: ${e.getMessage}. Check if your Haskell/Stack environment is working okay. Command was: $command")
        exit()
        null
    }
  }


  def exit(forceExit: Boolean = false): Unit = {
    if (!forceExit && !available) {
      println("Stack repl can not be stopped because it's already stopped or busy with stopping")
      return
    }

    available = false

    try {
      try {
        if (interoInputStream != null) {
          writeToIntero(":q")
        }
      }
      catch {
        case e: Exception =>
          println(s"Error while shutting down Stack repl. Error message: ${e.getMessage}")
      }
    } finally {
      if (procOut != null) {
        procOut.close()
      }
      if (procErr != null) {
        procErr.close()
      }
      if (interoInputStream != null) {
        try {
          interoInputStream.close()
        } catch {
          case _: Exception => ()
        }
      }
    }
    println("Stack repl stopped.")
  }

  def restart(): Unit = {
    exit()
    startIntero()
  }


  private def removePrompt(output: Seq[String]): Seq[String] = {
    if (output.isEmpty) {
      output
    } else {
      output.init
    }
  }

  private def convertOutputToOneMessagePerLine(output: Seq[String]) = {
    joinIndentedLines(output.filterNot(_.isEmpty))
  }

  private def joinIndentedLines(lines: Seq[String]): Seq[String] = {
    if (lines.size == 1) {
      lines
    } else {
      try {
        lines.foldLeft(ListBuffer[StringBuilder]())((lb, s) =>
          if (s.startsWith("  ")) {
            lb.last.append(s)
            lb
          }
          else {
            lb += new StringBuilder(2, s)
          }).map(_.toString)
      } catch {
        case _: NoSuchElementException =>
          System.out.println("Could not join indented lines")
          Seq()
      }
    }
  }

}

case class InteroOutput(stdOutLines: Seq[String] = Seq(), stdErrLines: Seq[String] = Seq())