|
@@ -118,6 +118,9 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
|
|
|
private val thread = new Thread {
|
|
private val thread = new Thread {
|
|
|
override def run() = {
|
|
override def run() = {
|
|
|
waitUntilReady()
|
|
waitUntilReady()
|
|
|
|
|
+
|
|
|
|
|
+ _state = Session.Idle()
|
|
|
|
|
+
|
|
|
loop()
|
|
loop()
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -132,10 +135,14 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
|
|
|
|
|
|
|
|
@tailrec
|
|
@tailrec
|
|
|
def loop(): Unit = {
|
|
def loop(): Unit = {
|
|
|
- _state = Session.Idle()
|
|
|
|
|
|
|
+ (_state, queue.take()) match {
|
|
|
|
|
+ case (Session.Error(), ExecuteRequest(code, promise)) =>
|
|
|
|
|
+ promise.failure(new Exception("session has been terminated"))
|
|
|
|
|
+ loop()
|
|
|
|
|
+
|
|
|
|
|
+ case (state, ExecuteRequest(code, promise)) =>
|
|
|
|
|
+ require(state == Session.Idle())
|
|
|
|
|
|
|
|
- queue.take() match {
|
|
|
|
|
- case ExecuteRequest(code, promise) =>
|
|
|
|
|
_state = Session.Busy()
|
|
_state = Session.Busy()
|
|
|
|
|
|
|
|
val msg = Map(
|
|
val msg = Map(
|
|
@@ -148,6 +155,7 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
|
|
|
val line = stdout.readLine()
|
|
val line = stdout.readLine()
|
|
|
// The python process shut down
|
|
// The python process shut down
|
|
|
if (line == null) {
|
|
if (line == null) {
|
|
|
|
|
+ _state = Session.Error()
|
|
|
promise.failure(new Exception("session has been terminated"))
|
|
promise.failure(new Exception("session has been terminated"))
|
|
|
} else {
|
|
} else {
|
|
|
val rep = parse(line)
|
|
val rep = parse(line)
|
|
@@ -156,12 +164,16 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
|
|
|
val content: JValue = rep \ "content"
|
|
val content: JValue = rep \ "content"
|
|
|
_history += content
|
|
_history += content
|
|
|
|
|
|
|
|
- promise.success(content)
|
|
|
|
|
|
|
+ _state = Session.Idle()
|
|
|
|
|
|
|
|
- loop()
|
|
|
|
|
|
|
+ promise.success(content)
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- case ShutdownRequest(promise) =>
|
|
|
|
|
|
|
+ loop()
|
|
|
|
|
+
|
|
|
|
|
+ case (_, ShutdownRequest(promise)) =>
|
|
|
|
|
+ require(state == Session.Idle() || state == Session.Error())
|
|
|
|
|
+
|
|
|
_state = Session.ShuttingDown()
|
|
_state = Session.ShuttingDown()
|
|
|
process.getInputStream.close()
|
|
process.getInputStream.close()
|
|
|
process.getOutputStream.close()
|
|
process.getOutputStream.close()
|
|
@@ -192,7 +204,7 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
|
|
|
promise.future
|
|
promise.future
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- override def close(): Future[Unit] = {
|
|
|
|
|
|
|
+ override def close(): Future[Unit] = synchronized {
|
|
|
_state match {
|
|
_state match {
|
|
|
case Session.ShutDown() =>
|
|
case Session.ShutDown() =>
|
|
|
Future.successful(())
|
|
Future.successful(())
|
|
@@ -202,14 +214,12 @@ private class PythonSession(process: Process, gatewayServer: GatewayServer) exte
|
|
|
Future.successful(())
|
|
Future.successful(())
|
|
|
}
|
|
}
|
|
|
case _ =>
|
|
case _ =>
|
|
|
- synchronized {
|
|
|
|
|
- val promise = Promise[Unit]()
|
|
|
|
|
- queue.put(ShutdownRequest(promise))
|
|
|
|
|
- promise.future.map({ case () =>
|
|
|
|
|
- thread.join()
|
|
|
|
|
- gatewayServer.shutdown()
|
|
|
|
|
- })
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ val promise = Promise[Unit]()
|
|
|
|
|
+ queue.put(ShutdownRequest(promise))
|
|
|
|
|
+ promise.future.map({ case () =>
|
|
|
|
|
+ thread.join()
|
|
|
|
|
+ gatewayServer.shutdown()
|
|
|
|
|
+ })
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|