Преглед изворни кода

[livy] Add support for paging through sessions

Erick Tryzelaar пре 10 година
родитељ
комит
c674916

+ 9 - 9
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/SessionManager.scala

@@ -18,13 +18,12 @@
 
 package com.cloudera.hue.livy.server
 
-import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicInteger
 
 import com.cloudera.hue.livy.Logging
 import org.json4s.JValue
 
-import scala.collection.convert.decorateAsScala._
+import scala.collection.mutable
 import scala.concurrent.{ExecutionContext, Future}
 
 object SessionManager {
@@ -43,12 +42,12 @@ class SessionManager[S <: Session](factory: SessionFactory[S])
   private implicit def executor: ExecutionContext = ExecutionContext.global
 
   protected[this] val _idCounter = new AtomicInteger()
-  protected[this] val _sessions = new ConcurrentHashMap[Int, S]().asScala
+  protected[this] val _sessions = mutable.Map[Int, S]()
 
   private val garbageCollector = new GarbageCollector
   garbageCollector.start()
 
-  def create(createRequest: JValue): Future[S] = {
+  def create(createRequest: JValue): Future[S] = synchronized {
     val id = _idCounter.getAndIncrement
     val session: Future[S] = factory.create(id, createRequest)
 
@@ -61,6 +60,8 @@ class SessionManager[S <: Session](factory: SessionFactory[S])
 
   def get(id: Int): Option[S] = _sessions.get(id)
 
+  def size(): Int = _sessions.size
+
   def all(): Iterable[S] = _sessions.values
 
   def delete(id: Int): Option[Future[Unit]] = {
@@ -69,15 +70,14 @@ class SessionManager[S <: Session](factory: SessionFactory[S])
 
   def delete(session: S): Future[Unit] = {
     session.stop().map { case _ =>
-      _sessions.remove(session.id)
+      synchronized {
+        _sessions.remove(session.id)
+      }
+
       Unit
     }
   }
 
-  def remove(id: Int): Option[S] = {
-    _sessions.remove(id)
-  }
-
   def shutdown(): Unit = {}
 
   def collectGarbage() = {

+ 10 - 2
apps/spark/java/livy-server/src/main/scala/com/cloudera/hue/livy/server/SessionServlet.scala

@@ -48,8 +48,16 @@ abstract class SessionServlet[S <: Session](sessionManager: SessionManager[S])
   }
 
   get("/") {
-    val sessions = sessionManager.all().map(serializeSession)
-    Map("sessions" -> sessions)
+    val from = params.get("from").map(_.toInt).getOrElse(0)
+    val size = params.get("size").map(_.toInt).getOrElse(100)
+
+    val sessions = sessionManager.all()
+
+    Map(
+      "from" -> from,
+      "total" -> sessionManager.size(),
+      "sessions" -> sessions.view(from, from + size).map(serializeSession)
+    )
   }
 
   val getSession = get("/:id") {