|
|
@@ -1,6 +1,6 @@
|
|
|
package com.cloudera.hue.livy.repl
|
|
|
|
|
|
-import java.io.{BufferedReader, PipedReader, PipedWriter, StringWriter}
|
|
|
+import java.io._
|
|
|
import java.util.concurrent.{BlockingQueue, SynchronousQueue}
|
|
|
|
|
|
import com.cloudera.hue.livy.{Complete, ExecuteResponse}
|
|
|
@@ -8,6 +8,7 @@ import org.apache.spark.repl.SparkILoop
|
|
|
|
|
|
import scala.annotation.tailrec
|
|
|
import scala.concurrent._
|
|
|
+import scala.concurrent.duration.Duration
|
|
|
import scala.tools.nsc.SparkHelper
|
|
|
import scala.tools.nsc.interpreter.{Formatting, _}
|
|
|
import scala.tools.nsc.util.ClassPath
|
|
|
@@ -15,14 +16,9 @@ import scala.tools.nsc.util.ClassPath
|
|
|
class SparkInterpreter {
|
|
|
private implicit def executor: ExecutionContext = ExecutionContext.global
|
|
|
|
|
|
- private val inQueue = new SynchronousQueue[Request]
|
|
|
- private val inWriter = new PipedWriter()
|
|
|
+ private val inQueue = new SynchronousQueue[ILoop.Request]
|
|
|
|
|
|
- org.apache.spark.repl.Main.interp = new ILoop(
|
|
|
- this,
|
|
|
- inQueue,
|
|
|
- new BufferedReader(new PipedReader(inWriter)),
|
|
|
- new StringWriter)
|
|
|
+ org.apache.spark.repl.Main.interp = new ILoop(inQueue)
|
|
|
|
|
|
// Launch the real interpreter thread.
|
|
|
private val thread = new Thread {
|
|
|
@@ -37,19 +33,40 @@ class SparkInterpreter {
|
|
|
org.apache.spark.repl.Main.interp.history.asStrings
|
|
|
}
|
|
|
|
|
|
- def execute(statement: String): Future[com.cloudera.hue.livy.ExecuteResponse] = {
|
|
|
- val promise = Promise[ExecuteResponse]()
|
|
|
- inQueue.put(ExecuteRequest(statement, promise))
|
|
|
- promise.future
|
|
|
+ def execute(statement: String): Future[ExecuteResponse] = {
|
|
|
+ val promise = Promise[ILoop.ExecuteResponse]()
|
|
|
+ inQueue.put(ILoop.ExecuteRequest(statement, promise))
|
|
|
+
|
|
|
+ for {
|
|
|
+ rep <- promise.future
|
|
|
+ } yield ExecuteResponse(0, List(statement), List(rep.output))
|
|
|
}
|
|
|
|
|
|
def close(): Unit = {
|
|
|
- inQueue.put(ShutdownRequest())
|
|
|
+ val promise = Promise[ILoop.ShutdownResponse]()
|
|
|
+ inQueue.put(ILoop.ShutdownRequest(promise))
|
|
|
+
|
|
|
+ Await.result(promise.future, Duration.Inf)
|
|
|
+
|
|
|
thread.join()
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-private class ILoop(parent: SparkInterpreter, inQueue: BlockingQueue[Request], in0: BufferedReader, outString: StringWriter) extends SparkILoop(in0, new JPrintWriter(outString)) {
|
|
|
+private object ILoop {
|
|
|
+ sealed trait Request
|
|
|
+ case class ExecuteRequest(statement: String, promise: Promise[ExecuteResponse]) extends Request
|
|
|
+ case class ShutdownRequest(promise: Promise[ShutdownResponse]) extends Request
|
|
|
+
|
|
|
+ case class ExecuteResponse(output: String)
|
|
|
+ case class ShutdownResponse()
|
|
|
+}
|
|
|
+
|
|
|
+// FIXME: The spark interpreter is written to own the event loop, so we need to invert it so we can inject our commands into it.
|
|
|
+private class ILoop(inQueue: BlockingQueue[ILoop.Request], outString: StringWriter = new StringWriter)
|
|
|
+ extends SparkILoop(
|
|
|
+ // we don't actually use the reader, so pass in a null reader for now.
|
|
|
+ new BufferedReader(new StringReader("")),
|
|
|
+ new JPrintWriter(outString)) {
|
|
|
|
|
|
class ILoopInterpreter extends SparkILoopInterpreter {
|
|
|
outer =>
|
|
|
@@ -104,32 +121,36 @@ private class ILoop(parent: SparkInterpreter, inQueue: BlockingQueue[Request], i
|
|
|
}
|
|
|
|
|
|
// return false if repl should exit
|
|
|
- def processLine(request: Request): Boolean = {
|
|
|
+ def processLine(request: ILoop.Request): Boolean = {
|
|
|
if (isAsync) {
|
|
|
if (!awaitInitialized()) return false
|
|
|
runThunks()
|
|
|
}
|
|
|
|
|
|
request match {
|
|
|
- case ExecuteRequest(statement, promise) =>
|
|
|
+ case ILoop.ExecuteRequest(statement, promise) =>
|
|
|
command(statement) match {
|
|
|
case Result(false, _) => false
|
|
|
case Result(true, finalLine) =>
|
|
|
finalLine match {
|
|
|
case Some(line) => addReplay(line)
|
|
|
- case _ =>
|
|
|
+ case None =>
|
|
|
}
|
|
|
|
|
|
- var output: String = outString.getBuffer.toString
|
|
|
+ var output = outString.getBuffer.toString
|
|
|
+
|
|
|
+ // Strip the trailing '\n'
|
|
|
output = output.substring(0, output.length - 1)
|
|
|
+
|
|
|
outString.getBuffer.setLength(0)
|
|
|
|
|
|
- val statement = ExecuteResponse(0, Complete(), List(), List(output))
|
|
|
- promise.success(statement)
|
|
|
+ promise.success(ILoop.ExecuteResponse(output))
|
|
|
|
|
|
true
|
|
|
}
|
|
|
- case ShutdownRequest() => false
|
|
|
+ case ILoop.ShutdownRequest(promise) =>
|
|
|
+ promise.success(ILoop.ShutdownResponse())
|
|
|
+ false
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -151,7 +172,3 @@ private class ILoop(parent: SparkInterpreter, inQueue: BlockingQueue[Request], i
|
|
|
innerLoop()
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
-private sealed trait Request
|
|
|
-private case class ExecuteRequest(statement: String, promise: Promise[ExecuteResponse]) extends Request
|
|
|
-private case class ShutdownRequest() extends Request
|