Refactoring search command to sttp

Author: Hariharan Ramanathan

src/main/scala/de/upb/cs/swt/delphi/cli/DelphiCLI.scala

@@ -49,7 +49,7 @@ object DelphiCLI {
         config.mode match {
           case "test" => TestCommand.execute
           case "retrieve" => RetrieveCommand.execute
-          //        case "search" => SearchCommand.execute(config)
+          case "search" => SearchCommand.execute
           case x => config.consoleOutput.outputError(s"Unknown command: $x")
         }
 

src/main/scala/de/upb/cs/swt/delphi/cli/commands/RetrieveCommand.scala

@@ -27,7 +27,7 @@ import scala.io.Source
   * The implementation of the retrieve command.
   * Retrieves the contents of the file at the endpoint specified by the config file, and prints them to stdout
   */
-object RetrieveCommand extends Command with DefaultJsonProtocol {
+object RetrieveCommand extends Command {
 
 
   override def execute(implicit config: Config, backend: SttpBackend[Id, Nothing]): Unit = {
@@ -45,7 +45,6 @@ object RetrieveCommand extends Command with DefaultJsonProtocol {
       }
     }
 
-    println(config.id)
     val result = executeGet(
       Seq("retrieve", checkTarget),
       Map("pretty" -> "")

src/main/scala/de/upb/cs/swt/delphi/cli/commands/SearchCommand.scala

@@ -16,124 +16,102 @@
 
 package de.upb.cs.swt.delphi.cli.commands
 
-import java.util.concurrent.{TimeUnit, TimeoutException}
-
-import akka.actor.ActorSystem
-import akka.http.scaladsl.Http
-import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
-import akka.http.scaladsl.marshalling.Marshal
-import akka.http.scaladsl.model._
-import akka.http.scaladsl.unmarshalling.Unmarshal
-import akka.stream.ActorMaterializer
-import akka.util.ByteString
-import de.upb.cs.swt.delphi.cli.Config
+import java.util.concurrent.TimeUnit
+
+import com.softwaremill.sttp._
+import com.softwaremill.sttp.sprayJson._
 import de.upb.cs.swt.delphi.cli.artifacts.SearchResult
-import de.upb.cs.swt.delphi.cli.artifacts.SearchResultJson._
-import spray.json.DefaultJsonProtocol
+import de.upb.cs.swt.delphi.cli.{Config, artifacts}
+import spray.json._
 
 import scala.concurrent.duration._
-import scala.concurrent.{Await, ExecutionContextExecutor, Future}
-import scala.util.{Failure, Success, Try}
 
-object SearchCommand extends Command with SprayJsonSupport with DefaultJsonProtocol {
+object SearchCommand extends Command with DefaultJsonProtocol{
 
-  val searchTimeout: Int = 10
+  val searchTimeout = 10.seconds
+  val TIMEOUT_CODE = 408
 
   /**
     * Executes the command implementation
     *
     * @param config The current configuration for the command
     */
-   def execute(config: Config)(implicit system: ActorSystem): Unit = {
-    implicit val ec = system.dispatcher
-    implicit val materializer = ActorMaterializer()
+  override def execute(implicit config: Config, backend: SttpBackend[Id, Nothing]): Unit = {
 
     def query = config.query
 
-    information(config)(s"Searching for artifacts matching ${'"'}$query${'"'}.")
-    val start = System.nanoTime()
+    information.apply(s"Searching for artifacts matching ${'"'}$query${'"'}.")
 
-    implicit val queryFormat = jsonFormat2(Query)
-    val baseUri = Uri(config.server)
-    val prettyParam = Map("pretty" -> "")
-    val searchUri = baseUri.withPath(baseUri.path + "/search").withQuery(akka.http.scaladsl.model.Uri.Query(prettyParam))
-    val responseFuture = Marshal(Query(query, config.limit)).to[RequestEntity] flatMap { entity =>
-      Http().singleRequest(HttpRequest(uri = searchUri, method = HttpMethods.POST, entity = entity))
-    }
 
-    Try(Await.result(responseFuture, Duration(config.timeout.getOrElse(searchTimeout) + " seconds"))).
-      map(response => parseResponse(response, config, start)(ec, materializer)).
-      recover {
-        case e : TimeoutException => {
-          error(config)("The query timed out after " + (System.nanoTime() - start).nanos.toUnit(TimeUnit.SECONDS) +
-            " seconds. To set a longer timeout, use the --timeout option.")
-          Failure(e)
-        }
-      }
+    val queryParams = Map("pretty" -> "")
+    val queryPayload: Query = Query(query,config.limit)
+    val searchUri = uri"${config.server}/search?$queryParams"
+
+    val request = sttp.body(queryPayload.toJson).post(searchUri)
+
+    val (res, time) = processRequest(request)
+    res.foreach(processResults(_, time))
   }
 
-  private def parseResponse(response: HttpResponse, config: Config, start: Long)
-                           (implicit ec: ExecutionContextExecutor, materializer: ActorMaterializer): Unit = {
+  private def processRequest(req: Request[String, Nothing])
+                            (implicit config: Config,
+                             backend: SttpBackend[Id, Nothing]): (Option[String], FiniteDuration) = {
+    val start = System.nanoTime()
+    val res: Id[Response[String]] = req.readTimeout(searchTimeout).send()
+    val end = System.nanoTime()
+    val took = (end - start).nanos
 
-    val resultFuture: Future[String] = response match {
-      case HttpResponse(StatusCodes.OK, headers, entity, _) =>
-        entity.dataBytes.runFold(ByteString(""))(_ ++ _).map { body =>
-          body.utf8String
-        }
-      case resp@HttpResponse(code, _, _, _) => {
-        error(config)("Request failed, response code: " + code)
-        resp.discardEntityBytes()
-        Future("")
-      }
-    }
+    if (res.code == TIMEOUT_CODE) {
 
-    val result = Await.result(resultFuture, Duration.Inf)
+      error.apply(s"The query timed out after   ${took.toSeconds} seconds. " +
+        "To set a longer timeout, use the --timeout option.")
+    }
+    val resStr = res.body match {
+      case Left(v) =>
+        error.apply(s"Search request failed \n $v")
+        println(v)
+        None
+      case Right(v) =>
+        Some(v)
+    }
+    (resStr, took)
+  }
 
-    val took = (System.nanoTime() - start).nanos.toUnit(TimeUnit.SECONDS)
+  private def processResults(res: String, queryRuntime: FiniteDuration)(implicit config: Config) = {
 
-    if (config.raw || result.equals("")) {
-      reportResult(config)(result)
+    if (config.raw || res.equals("")) {
+      reportResult.apply(res)
+    }
+    if (!(config.raw || res.equals("")) || !config.csv.equals("")) {
+      import artifacts.SearchResultJson._
+      val jsonArr = res.parseJson.asInstanceOf[JsArray].elements
+      val retrieveResults = jsonArr.map(r => r.convertTo[SearchResult]).toList
+      onProperSearchResults(retrieveResults)
     }
 
-    if(!(config.raw || result.equals("")) || !config.csv.equals("")) {
-      val unmarshalledFuture = Unmarshal(result).to[List[SearchResult]]
+    def onProperSearchResults(sr: List[SearchResult]) = {
 
-      val processFuture = unmarshalledFuture.transform {
-        case Success(unmarshalled) => {
-          processResults(config, unmarshalled, took)
-          Success(unmarshalled)
-        }
-        case Failure(e) => {
-          error(config)(result)
-          Failure(e)
+      val capMessage = {
+        config.limit match {
+          case Some(limit) if (limit <= sr.size)
+          => s"Results may be capped by result limit set to $limit."
+          case None if (sr.size >= 50)
+          => "Results may be capped by default limit of 50 returned results. Use --limit to extend the result set."
+          case _
+          => ""
         }
       }
-    }
-  }
 
-  private def processResults(config: Config, results: List[SearchResult], queryRuntime: Double) = {
-    val capMessage = {
-      config.limit match {
-        case Some(limit) if (limit <= results.size)
-        => s"Results may be capped by result limit set to $limit."
-        case None if (results.size >= 50)
-        => "Results may be capped by default limit of 50 returned results. Use --limit to extend the result set."
-        case _
-        => ""
-      }
-    }
-    success(config)(s"Found ${results.size} item(s). $capMessage")
-    reportResult(config)(results)
+      success.apply(s"Found ${sr.size} item(s). $capMessage")
+      reportResult.apply(sr)
 
-    information(config)(f"Query took $queryRuntime%.2fs.")
+      information.apply(f"Query took ${queryRuntime.toUnit(TimeUnit.SECONDS)}%.2fs.")
 
-    if(!config.csv.equals("")) {
-      exportResult(config)(results)
-      information(config)("Results written to file '" + config.csv + "'")
+      if (!config.csv.equals("")) {
+        exportResult.apply(sr)
+        information.apply("Results written to file '" + config.csv + "'")
+      }
     }
   }
 
-  case class Query(query: String,
-                   limit: Option[Int] = None)
-
 }

src/main/scala/de/upb/cs/swt/delphi/cli/commands/package.scala

@@ -0,0 +1,28 @@
+// Copyright (C) 2018 The Delphi Team.
+// See the LICENCE file distributed with this work for additional
+// information regarding copyright ownership.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package de.upb.cs.swt.delphi.cli
+
+import spray.json._
+
+package object commands extends DefaultJsonProtocol {
+
+
+  case class Query(query: String,
+                   limit: Option[Int] = None)
+
+  implicit val queryFormat = jsonFormat2(Query)
+
+}