|
@@ -6,6 +6,7 @@ import org.json4s.jackson.Serialization.write
|
|
|
import org.json4s.{DefaultFormats, Formats}
|
|
import org.json4s.{DefaultFormats, Formats}
|
|
|
|
|
|
|
|
import scala.annotation.tailrec
|
|
import scala.annotation.tailrec
|
|
|
|
|
+import scala.collection.mutable.ArrayBuffer
|
|
|
import scala.concurrent.{Future, _}
|
|
import scala.concurrent.{Future, _}
|
|
|
|
|
|
|
|
abstract class SparkWebSession(val id: String, hostname: String, port: Int) extends Session with Logging {
|
|
abstract class SparkWebSession(val id: String, hostname: String, port: Int) extends Session with Logging {
|
|
@@ -17,54 +18,77 @@ abstract class SparkWebSession(val id: String, hostname: String, port: Int) exte
|
|
|
private[this] var _state: State = Idle()
|
|
private[this] var _state: State = Idle()
|
|
|
private[this] val svc = host(hostname, port)
|
|
private[this] val svc = host(hostname, port)
|
|
|
|
|
|
|
|
|
|
+ private[this] var executedStatements = 0
|
|
|
|
|
+ private[this] var statements_ = new ArrayBuffer[ExecuteResponse]
|
|
|
|
|
+
|
|
|
override def lastActivity: Long = _lastActivity
|
|
override def lastActivity: Long = _lastActivity
|
|
|
|
|
|
|
|
override def state: State = _state
|
|
override def state: State = _state
|
|
|
|
|
|
|
|
- override def executeStatement(statement: String): Future[ExecuteResponse] = {
|
|
|
|
|
|
|
+ override def executeStatement(statement: String): (Int, Future[ExecuteResponse]) = {
|
|
|
ensureIdle {
|
|
ensureIdle {
|
|
|
_state = Busy()
|
|
_state = Busy()
|
|
|
touchLastActivity()
|
|
touchLastActivity()
|
|
|
|
|
|
|
|
|
|
+ executedStatements += 1
|
|
|
|
|
+
|
|
|
var req = (svc / "statements").setContentType("application/json", "UTF-8")
|
|
var req = (svc / "statements").setContentType("application/json", "UTF-8")
|
|
|
req = req << write(ExecuteRequest(statement))
|
|
req = req << write(ExecuteRequest(statement))
|
|
|
|
|
|
|
|
- Http(req OK as.json4s.Json).map {
|
|
|
|
|
- transition(Idle())
|
|
|
|
|
- _.extract[ExecuteResponse]
|
|
|
|
|
|
|
+ val future = Http(req OK as.json4s.Json).map { case (resp) =>
|
|
|
|
|
+ synchronized {
|
|
|
|
|
+ transition(Idle())
|
|
|
|
|
+ val response: ExecuteResponse = resp.extract[ExecuteResponse]
|
|
|
|
|
+ statements_ += response
|
|
|
|
|
+ response
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+ (executedStatements - 1, future)
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
override def statement(statementId: Int): Future[ExecuteResponse] = {
|
|
override def statement(statementId: Int): Future[ExecuteResponse] = {
|
|
|
ensureRunning {
|
|
ensureRunning {
|
|
|
- val req = svc / "statements" / statementId
|
|
|
|
|
|
|
+ if (statementId < statements_.length) {
|
|
|
|
|
+ Future.successful(statements_(statementId))
|
|
|
|
|
+ } else {
|
|
|
|
|
+ val req = svc / "statements" / statementId
|
|
|
|
|
|
|
|
- for {
|
|
|
|
|
- body <- Http(req OK as.json4s.Json)
|
|
|
|
|
- } yield body.extract[ExecuteResponse]
|
|
|
|
|
|
|
+ for {
|
|
|
|
|
+ body <- Http(req OK as.json4s.Json)
|
|
|
|
|
+ } yield body.extract[ExecuteResponse]
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
override def statements(): Future[List[ExecuteResponse]] = {
|
|
override def statements(): Future[List[ExecuteResponse]] = {
|
|
|
ensureRunning {
|
|
ensureRunning {
|
|
|
- val req = svc / "statements"
|
|
|
|
|
|
|
+ if (_state == Idle()) {
|
|
|
|
|
+ Future.successful(statements_.toList)
|
|
|
|
|
+ } else {
|
|
|
|
|
+ val req = svc / "statements"
|
|
|
|
|
|
|
|
- for {
|
|
|
|
|
- body <- Http(req OK as.json4s.Json)
|
|
|
|
|
- } yield body.extract[List[ExecuteResponse]]
|
|
|
|
|
|
|
+ for {
|
|
|
|
|
+ body <- Http(req OK as.json4s.Json)
|
|
|
|
|
+ } yield body.extract[List[ExecuteResponse]]
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
override def statements(fromIndex: Integer, toIndex: Integer): Future[List[ExecuteResponse]] = {
|
|
override def statements(fromIndex: Integer, toIndex: Integer): Future[List[ExecuteResponse]] = {
|
|
|
ensureRunning {
|
|
ensureRunning {
|
|
|
- val req = (svc / "statements")
|
|
|
|
|
- .addQueryParameter("from", fromIndex.toString)
|
|
|
|
|
- .addQueryParameter("to", toIndex.toString)
|
|
|
|
|
|
|
+ if (_state == Idle()) {
|
|
|
|
|
+ Future.successful(statements_.slice(fromIndex, toIndex).toList)
|
|
|
|
|
+ } else {
|
|
|
|
|
+ val req = (svc / "statements")
|
|
|
|
|
+ .addQueryParameter("from", fromIndex.toString)
|
|
|
|
|
+ .addQueryParameter("to", toIndex.toString)
|
|
|
|
|
|
|
|
- for {
|
|
|
|
|
- body <- Http(req OK as.json4s.Json)
|
|
|
|
|
- } yield body.extract[List[ExecuteResponse]]
|
|
|
|
|
|
|
+ for {
|
|
|
|
|
+ body <- Http(req OK as.json4s.Json)
|
|
|
|
|
+ } yield body.extract[List[ExecuteResponse]]
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
override def interrupt(): Future[Unit] = {
|
|
override def interrupt(): Future[Unit] = {
|