- Added an execution context to the PoolingRestClient to be able to propagate it correctly from custom implementations

Author: YevSent

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/AkkaContainerClient.scala

@@ -70,8 +70,8 @@ protected class AkkaContainerClient(
   port: Int,
   timeout: FiniteDuration,
   queueSize: Int,
-  retryInterval: FiniteDuration = 100.milliseconds)(implicit logging: Logging, as: ActorSystem)
-    extends PoolingRestClient("http", hostname, port, queueSize, timeout = Some(timeout))
+  retryInterval: FiniteDuration = 100.milliseconds)(implicit logging: Logging, as: ActorSystem, ec: ExecutionContext)
+    extends PoolingRestClient("http", hostname, port, queueSize, timeout = Some(timeout))(as, ec)
     with ContainerClient {
 
   def close() = shutdown()

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/ElasticSearchLogStore.scala

@@ -79,7 +79,7 @@ class ElasticSearchLogStore(
     elasticSearchConfig.protocol,
     elasticSearchConfig.host,
     elasticSearchConfig.port,
-    httpFlow)
+    httpFlow)(system, system.dispatcher)
 
   private def transcribeLogs(queryResult: EsSearchResult): ActivationLogs =
     ActivationLogs(queryResult.hits.hits.map(_.source.convertTo[UserLogEntry].toFormattedString))

common/scala/src/main/scala/org/apache/openwhisk/core/containerpool/logging/ElasticSearchRestClient.scala

@@ -17,20 +17,16 @@
 
 package org.apache.openwhisk.core.containerpool.logging
 
-import scala.concurrent.Future
+import scala.concurrent.{ExecutionContext, Future, Promise}
 import scala.util.{Either, Try}
-
 import akka.actor.ActorSystem
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.model.HttpMethods.{GET, POST}
 import akka.http.scaladsl.model.headers.Accept
 import akka.stream.scaladsl.Flow
 
-import scala.concurrent.Promise
 import scala.util.Try
-
 import spray.json._
-
 import org.apache.openwhisk.http.PoolingRestClient
 import org.apache.openwhisk.http.PoolingRestClient._
 
@@ -154,8 +150,9 @@ class ElasticSearchRestClient(
   host: String,
   port: Int,
   httpFlow: Option[Flow[(HttpRequest, Promise[HttpResponse]), (Try[HttpResponse], Promise[HttpResponse]), Any]] = None)(
-  implicit system: ActorSystem)
-    extends PoolingRestClient(protocol, host, port, 16 * 1024, httpFlow) {
+  implicit system: ActorSystem,
+  ec: ExecutionContext)
+    extends PoolingRestClient(protocol, host, port, 16 * 1024, httpFlow)(system, ec) {
 
   import ElasticSearchJsonProtocol._
 

common/scala/src/main/scala/org/apache/openwhisk/core/database/CouchDbRestClient.scala

@@ -42,9 +42,11 @@ import scala.concurrent.{ExecutionContext, Future}
 class CouchDbRestClient(protocol: String, host: String, port: Int, username: String, password: String, db: String)(
   implicit system: ActorSystem,
   logging: Logging)
-    extends PoolingRestClient(protocol, host, port, 16 * 1024) {
+    extends PoolingRestClient(protocol, host, port, 16 * 1024)(
+      system,
+      system.dispatchers.lookup("dispatchers.couch-dispatcher")) {
 
-  protected implicit override val context: ExecutionContext = system.dispatchers.lookup("dispatchers.couch-dispatcher")
+  protected implicit val context: ExecutionContext = system.dispatchers.lookup("dispatchers.couch-dispatcher")
 
   // Headers common to all requests.
   protected val baseHeaders: List[HttpHeader] =

common/scala/src/main/scala/org/apache/openwhisk/http/PoolingRestClient.scala

@@ -47,13 +47,11 @@ class PoolingRestClient(
   port: Int,
   queueSize: Int,
   httpFlow: Option[Flow[(HttpRequest, Promise[HttpResponse]), (Try[HttpResponse], Promise[HttpResponse]), Any]] = None,
-  timeout: Option[FiniteDuration] = None)(implicit system: ActorSystem) {
+  timeout: Option[FiniteDuration] = None)(implicit system: ActorSystem, ec: ExecutionContext) {
   require(protocol == "http" || protocol == "https", "Protocol must be one of { http, https }.")
 
   private val logging = new AkkaLogging(system.log)
 
-  protected implicit val context: ExecutionContext = system.dispatcher
-
   //if specified, override the ClientConnection idle-timeout and keepalive socket option value
   private val timeoutSettings = {
     val cps = ConnectionPoolSettings(system.settings.config)
@@ -137,7 +135,7 @@ class PoolingRestClient(
   def shutdown(): Future[Unit] = {
     killSwitch.shutdown()
     Try(requestQueue.complete()).recover {
-      case t: IllegalStateException => logging.error(this, t.getMessage)
+      case t: IllegalStateException => logging.warn(this, t.getMessage)
     }
     Future.unit
   }