Add tasks status definition + use HashedWheelTimer in waitFor

src/main/scala/algolia/AlgoliaDsl.scala

@@ -53,6 +53,7 @@ trait AlgoliaDsl
     with SearchDsl
     with SynonymsDsl
     with WaitForTaskDsl
+    with TaskStatusDsl
 
 object AlgoliaDsl extends AlgoliaDsl {
 

src/main/scala/algolia/definitions/TaskStatusDefinition.scala

@@ -0,0 +1,64 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2016 Algolia
+ * http://www.algolia.com/
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+package algolia.definitions
+
+import algolia.http.{GET, HttpPayload}
+import algolia.responses.{AlgoliaTask, TaskStatus}
+import algolia.{AlgoliaClient, Executable}
+
+import scala.concurrent.{ExecutionContext, Future}
+
+case class TaskStatusDefinition(taskId: Long, index: Option[String] = None) extends Definition {
+
+  def from(index: String): TaskStatusDefinition = copy(index = Some(index))
+
+  override private[algolia] def build(): HttpPayload = {
+    HttpPayload(
+      GET,
+      Seq("1", "indexes") ++ index ++ Seq("task", taskId.toString),
+      isSearch = true
+    )
+  }
+
+}
+
+trait TaskStatusDsl {
+
+  case object getStatus {
+    def task(task: AlgoliaTask): TaskStatusDefinition =
+      TaskStatusDefinition(task.idToWaitFor())
+  }
+
+  implicit object TaskStatusExecutable extends Executable[TaskStatusDefinition, TaskStatus] {
+
+    override def apply(client: AlgoliaClient, query: TaskStatusDefinition)(
+        implicit executor: ExecutionContext): Future[TaskStatus] = {
+      client.request[TaskStatus](query.build())
+    }
+
+  }
+
+}

src/main/scala/algolia/definitions/WaitForTaskDefinition.scala

@@ -25,11 +25,12 @@
 
 package algolia.definitions
 
-import java.util.{Timer, TimerTask}
+import java.util.concurrent.{Executors, ThreadFactory, TimeUnit}
 
-import algolia.http.{GET, HttpPayload}
+import algolia.http.HttpPayload
 import algolia.responses.{AlgoliaTask, TaskStatus}
 import algolia.{AlgoliaClient, Executable}
+import io.netty.util.{HashedWheelTimer, Timeout, TimerTask}
 
 import scala.concurrent.{ExecutionContext, Future, Promise}
 
@@ -45,13 +46,8 @@ case class WaitForTaskDefinition(taskId: Long,
 
   def maxDelay(delay: Long): WaitForTaskDefinition = copy(maxDelay = delay)
 
-  override private[algolia] def build(): HttpPayload = {
-    HttpPayload(
-      GET,
-      Seq("1", "indexes") ++ index ++ Seq("task", taskId.toString),
-      isSearch = true
-    )
-  }
+  override private[algolia] def build(): HttpPayload = TaskStatusDefinition(taskId, index).build()
+
 }
 
 trait WaitForTaskDsl {
@@ -64,6 +60,19 @@ trait WaitForTaskDsl {
   implicit object WaitForTaskDefinitionExecutable
       extends Executable[WaitForTaskDefinition, TaskStatus] {
 
+    // Run every 50 ms, use a wheel with 512 buckets
+    private lazy val timer = {
+      val threadFactory = new ThreadFactory {
+        override def newThread(r: Runnable): Thread = {
+          val t = Executors.defaultThreadFactory().newThread(r)
+          t.setDaemon(true)
+          t.setName("algolia-waitfor-thread")
+          t
+        }
+      }
+      new HashedWheelTimer(threadFactory, 50, TimeUnit.MILLISECONDS, 512)
+    }
+
     /**
       * Wait for the completion of a task
       * By default it waits `baseDelay` and multiple it by 2 each time until it's greater than `maxDelay`:
@@ -79,32 +88,30 @@ trait WaitForTaskDsl {
       */
     override def apply(client: AlgoliaClient, query: WaitForTaskDefinition)(
         implicit executor: ExecutionContext): Future[TaskStatus] = {
-      def request(d: Long): Future[TaskStatus] =
+
+      def request(d: Long, totalDelay: Long): Future[TaskStatus] =
         delay[TaskStatus](d) {
           client.request[TaskStatus](query.build())
         }.flatMap { res =>
           if (res.status == "published") {
             Future.successful(res)
-          } else if (d > query.maxDelay) {
+          } else if (totalDelay > query.maxDelay) {
             Future.failed(WaitForTimeoutException(
               s"Waiting for task `${query.taskId}` on index `${query.index.get}` timeout after ${d}ms"))
           } else {
-            request(d * 2)
+            request(d * 2, totalDelay + d)
           }
         }
 
-      request(query.baseDelay)
+      request(query.baseDelay, 0L)
     }
 
-    //from http://stackoverflow.com/questions/16359849/scala-scheduledfuture
     private def delay[T](delay: Long)(block: => Future[T]): Future[T] = {
       val promise = Promise[T]()
-      val t = new Timer()
-      t.schedule(new TimerTask {
-        override def run(): Unit = {
-          promise.completeWith(block)
-        }
-      }, delay)
+      val task = new TimerTask {
+        override def run(timeout: Timeout): Unit = promise.completeWith(block)
+      }
+      timer.newTimeout(task, delay, TimeUnit.MILLISECONDS)
       promise.future
     }
   }

src/test/scala/algolia/AlgoliaTest.scala

@@ -29,7 +29,7 @@ import algolia.AlgoliaDsl._
 import algolia.responses.{AlgoliaTask, TasksMultipleIndex}
 import org.scalamock.scalatest.MockFactory
 import org.scalatest._
-import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.concurrent.{Eventually, ScalaFutures}
 import org.scalatest.time.{Millis, Seconds, Span}
 
 import scala.concurrent.{ExecutionContext, Future}
@@ -43,7 +43,8 @@ class AlgoliaTest
     with ScalaFutures
     with Inside
     with MockFactory
-    with EitherValues {
+    with EitherValues
+    with Eventually {
 
   val applicationId: String = System.getenv("APPLICATION_ID")
   val apiKey: String = System.getenv("API_KEY")
@@ -61,12 +62,16 @@ class AlgoliaTest
     client.close()
   }
 
-  def taskShouldBeCreatedAndWaitForIt(task: Future[AlgoliaTask], index: String)(
-      implicit ec: ExecutionContext): Unit = {
-    val t: AlgoliaTask = whenReady(task) { result =>
+  def taskShouldBeCreated(task: Future[AlgoliaTask])(implicit ec: ExecutionContext): AlgoliaTask = {
+    whenReady(task) { result =>
       result.idToWaitFor should not be 0
-      result //for getting it after
+      result
     }
+  }
+
+  def taskShouldBeCreatedAndWaitForIt(task: Future[AlgoliaTask], index: String)(
+      implicit ec: ExecutionContext): Unit = {
+    val t: AlgoliaTask = taskShouldBeCreated(task)
 
     val waiting = client.execute {
       waitFor task t from index maxDelay (60 * 10 * 1000) //600 seconds

src/test/scala/algolia/dsl/TaskStatusTest.scala

@@ -0,0 +1,52 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2016 Algolia
+ * http://www.algolia.com/
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+package algolia.dsl
+
+import algolia.AlgoliaDsl._
+import algolia.AlgoliaTest
+import algolia.http.{GET, HttpPayload}
+import algolia.responses.Task
+
+class TaskStatusTest extends AlgoliaTest {
+
+  describe("get task status") {
+
+    it("get the status of a task") {
+      getStatus task Task(1L, None) from "toto"
+    }
+
+    it("should call API") {
+      (getStatus task Task(1L, None) from "toto").build() should be(
+        HttpPayload(
+          GET,
+          Seq("1", "indexes", "toto", "task", "1"),
+          isSearch = true
+        )
+      )
+    }
+  }
+
+}

src/test/scala/algolia/integration/TaskStatusIntegrationTest.scala

@@ -0,0 +1,73 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2016 Algolia
+ * http://www.algolia.com/
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+
+package algolia.integration
+
+import algolia.AlgoliaDsl._
+import algolia.AlgoliaTest
+import algolia.responses.{TaskIndexing, TasksMultipleIndex}
+
+import scala.concurrent.Future
+
+class TaskStatusIntegrationTest extends AlgoliaTest {
+
+  after {
+    clearIndices("indexToGetTaskStatus")
+  }
+
+  it("should get the status of a single object task") {
+    val create: Future[TaskIndexing] = client.execute {
+      index into "indexToGetTaskStatus" `object` ObjectToGet("1", "toto")
+    }
+
+    val task = taskShouldBeCreated(create)
+
+    eventually {
+      val status = client.execute { getStatus task task from "indexToGetTaskStatus" }
+      whenReady(status) { result =>
+        result.status == "published"
+      }
+    }
+  }
+
+  it("should get the status of a multiple objects task") {
+    val create: Future[TasksMultipleIndex] = client.execute {
+      batch(
+        index into "indexToGetTaskStatus" `object` ObjectToGet("1", "toto"),
+        index into "indexToGetTaskStatus" `object` ObjectToGet("2", "tata")
+      )
+    }
+
+    val task = taskShouldBeCreated(create)
+
+    eventually {
+      val status = client.execute { getStatus task task from "indexToGetTaskStatus" }
+      whenReady(status) { result =>
+        result.status == "published"
+      }
+    }
+  }
+
+}