Rearrange ExecutionContexts

Author: rtyley

aws-s3/aws-sdk-v2/src/main/scala/com/gu/etagcaching/aws/sdkv2/s3/S3ObjectFetching.scala

@@ -15,13 +15,15 @@ import scala.concurrent.{ExecutionContext, Future}
 
 case class S3ObjectFetching[Response](s3Client: S3AsyncClient, transformer: Transformer[Response])
   extends Fetching[ObjectId, Response] {
+  implicit val ec: ExecutionContext = ExecutionContext.parasitic // Only if suitable...
+
   private def wrapWithETag(resp: Response): ETaggedData[Response] =
     ETaggedData(transformer.rawResponseObjectOf(resp).eTag(), resp)
 
   private def performFetch(
     resourceId: ObjectId,
     reqModifier: Endo[GetObjectRequest.Builder] = identity,
-  )(implicit ec: ExecutionContext): Future[MissingOrETagged[Response]] = {
+  ): Future[MissingOrETagged[Response]] = {
     val requestBuilder = GetObjectRequest.builder().bucket(resourceId.bucket).key(resourceId.key)
     val request = reqModifier(requestBuilder).build()
     s3Client.getObject(request, transformer.asyncResponseTransformer())
@@ -35,9 +37,9 @@ case class S3ObjectFetching[Response](s3Client: S3AsyncClient, transformer: Tran
       }
   }
 
-  def fetch(key: ObjectId)(implicit ec: ExecutionContext): Future[MissingOrETagged[Response]] = performFetch(key)
+  def fetch(key: ObjectId): Future[MissingOrETagged[Response]] = performFetch(key)
 
-  def fetchOnlyIfETagChanged(key: ObjectId, eTag: String)(implicit ec: ExecutionContext): Future[Option[MissingOrETagged[Response]]] =
+  def fetchOnlyIfETagChanged(key: ObjectId, eTag: String): Future[Option[MissingOrETagged[Response]]] =
     performFetch(key, _.ifNoneMatch(eTag)).map(Some(_)).recover {
       case e: S3Exception if e.statusCode == HTTP_NOT_MODIFIED => None // no fresh download because the ETag matched!
     }
@@ -48,5 +50,5 @@ object S3ObjectFetching {
    * Convenience method for creating a fetcher that just returns a simple byte array.
    */
   def byteArraysWith(s3AsyncClient: S3AsyncClient): S3ByteArrayFetching =
-    S3ObjectFetching(s3AsyncClient, Bytes).mapResponse(_.asByteArray())
+    S3ObjectFetching(s3AsyncClient, Bytes).mapResponse(_.asByteArray())(ExecutionContext.parasitic)
 }

core/src/main/scala/com/gu/etagcaching/ETagCache.scala

@@ -17,7 +17,7 @@ import scala.concurrent.{ExecutionContext, Future}
  * ETagCache caches resolved-key-values along with their `ETag`, a content hash supplied & recognised by the
  * remote service. ETagCache (when used with a [[AlwaysWaitForRefreshedValue]] policy)
  * ''will'' always connect to the service for each request for a key-value, but where it already holds a
- * cached `ETaggedData`, it will send the `ETag` with it's request, and the service will return a
+ * cached `ETaggedData`, it will send the `ETag` with its request, and the service will return a
  * response indicating if the content has changed or not. If the content is unchanged, the response will
  * be blank, saving network bandwidth, and the old value can be used, saving CPU by not having to parse the
  * data again.
@@ -32,18 +32,17 @@ class ETagCache[K, V](
   loading: Loading[K, V],
   freshnessPolicy: FreshnessPolicy,
   configureCache: ConfigCache
-)(implicit ec: ExecutionContext) {
-
+) {
   private val cache: AsyncLoadingCache[K, MissingOrETagged[V]] = configureCache(Scaffeine()).buildAsyncFuture[K, MissingOrETagged[V]](
     loader = loading.fetchAndParse,
     reloadLoader = Some(
        (key: K, old: MissingOrETagged[V]) => old match {
          case Missing => loading.fetchAndParse(key)
-         case oldETaggedData: ETaggedData[V] => loading.fetchThenParseIfNecessary(key, oldETaggedData)
+         case oldETaggedData: ETaggedData[V] @unchecked => loading.fetchThenParseIfNecessary(key, oldETaggedData)
        }
     ))
 
   private val read = freshnessPolicy.on(cache)
 
-  def get(key: K): Future[Option[V]] = read(key).map(_.toOption)
+  def get(key: K): Future[Option[V]] = read(key).map(_.toOption)(ExecutionContext.parasitic)
 }

core/src/main/scala/com/gu/etagcaching/Loading.scala

@@ -18,27 +18,27 @@ import scala.concurrent.{ExecutionContext, Future}
  * @tparam V The 'value' for the key - a parsed representation of whatever was in the resource data.
  */
 trait Loading[K, V] {
-  def fetchAndParse(k: K)(implicit ec: ExecutionContext): Future[MissingOrETagged[V]]
+  def fetchAndParse(k: K): Future[MissingOrETagged[V]]
 
   /**
    * When we have ''old'' `ETaggedData`, we can send the `ETag` with the Fetch request, and the server will return
    * a blank HTTP 304 `Not Modified` response if the content hasn't changed - which means we DO NOT need to parse
    * any new data, and can just reuse our old data, saving us CPU time and network bandwidth.
    */
-  def fetchThenParseIfNecessary(k: K, oldV: ETaggedData[V])(implicit ec: ExecutionContext): Future[MissingOrETagged[V]]
+  def fetchThenParseIfNecessary(k: K, oldV: ETaggedData[V]): Future[MissingOrETagged[V]]
 
   /**
    * Add a handler for doing side-effectful logging of updates.
    */
-  def onUpdate(handler: Update[K,V] => Unit): Loading[K, V] = OnUpdate(this)(handler)
+  def onUpdate(handler: Update[K,V] => Unit)(implicit ec: ExecutionContext): Loading[K, V] = OnUpdate(this)(handler)
 }
 
 object Loading {
-  def by[K, Response, V](fetching: Fetching[K, Response])(parse: Response => V): Loading[K, V] = new Loading[K, V] {
-    def fetchAndParse(key: K)(implicit ec: ExecutionContext): Future[MissingOrETagged[V]] =
+  def by[K, Response, V](fetching: Fetching[K, Response])(parse: Response => V)(implicit parsingEC: ExecutionContext): Loading[K, V] = new Loading[K, V] {
+    def fetchAndParse(key: K): Future[MissingOrETagged[V]] =
       fetching.fetch(key).map(_.map(parse))
 
-    def fetchThenParseIfNecessary(key: K, oldV: ETaggedData[V])(implicit ec: ExecutionContext): Future[MissingOrETagged[V]] =
+    def fetchThenParseIfNecessary(key: K, oldV: ETaggedData[V]): Future[MissingOrETagged[V]] =
       fetching.fetchOnlyIfETagChanged(key, oldV.eTag).map {
         case None => oldV // we got HTTP 304 'NOT MODIFIED': there's no new data - old data is still valid
         case Some(freshResponse) => freshResponse.map(parse)
@@ -52,17 +52,19 @@ object Loading {
 
   /**
    * Wrapper round an underlying instance of Loading which adds handler for doing side-effectful logging of updates.
+   *
+   * @param loggingEC ExecutionContext for the `handler` function to run in
    */
-  case class OnUpdate[K, V](underlying: Loading[K, V])(handler: Update[K,V] => Unit)
+  case class OnUpdate[K, V](underlying: Loading[K, V])(handler: Update[K,V] => Unit)(implicit loggingEC: ExecutionContext)
     extends Loading[K, V] {
 
-    override def fetchAndParse(key: K)(implicit ec: ExecutionContext): Future[MissingOrETagged[V]] =
+    override def fetchAndParse(key: K): Future[MissingOrETagged[V]] =
       handle(key, None, underlying.fetchAndParse(key))
 
-    override def fetchThenParseIfNecessary(key: K, oldV: ETaggedData[V])(implicit ec: ExecutionContext): Future[MissingOrETagged[V]] =
+    override def fetchThenParseIfNecessary(key: K, oldV: ETaggedData[V]): Future[MissingOrETagged[V]] =
       handle(key, oldV.toOption, underlying.fetchThenParseIfNecessary(key, oldV))
 
-    private def handle(key: K, oldV: Option[V], fut: Future[MissingOrETagged[V]])(implicit ec: ExecutionContext): Future[MissingOrETagged[V]] = {
+    private def handle(key: K, oldV: Option[V], fut: Future[MissingOrETagged[V]]): Future[MissingOrETagged[V]] = {
       for {
         wrappedNewV <- fut
       } {

core/src/main/scala/com/gu/etagcaching/fetching/Fetching.scala

@@ -9,33 +9,35 @@ import scala.concurrent.{ExecutionContext, Future}
 import scala.util.Try
 
 trait Fetching[K, Response] {
-  def fetch(key: K)(implicit ec: ExecutionContext): Future[MissingOrETagged[Response]]
+  def fetch(key: K): Future[MissingOrETagged[Response]]
 
-  def fetchOnlyIfETagChanged(key: K, eTag: String)(implicit ec: ExecutionContext): Future[Option[MissingOrETagged[Response]]]
+  def fetchOnlyIfETagChanged(key: K, eTag: String): Future[Option[MissingOrETagged[Response]]]
 
+  /**
+   * @param recordingEC the ExecutionContext used for
+   */
   def timing(
     attemptWith: Duration => Unit = _ => (),
     successWith: Duration => Unit = _ => (),
     fullFetchWith: Duration => Unit = _ => (),
     notModifiedWith: Duration => Unit = _ => ()
-  ): Fetching[K, Response] =
-    DurationRecorder(this) { recorded =>
-      val duration = recorded.duration
-      attemptWith(duration)
-      recorded.result.foreach { successfulResponse =>
-        successWith(duration)
-        successfulResponse match {
-          case FullFetch => fullFetchWith(duration)
-          case NotModified => notModifiedWith(duration)
-        }
+  )(implicit recordingEC: ExecutionContext): Fetching[K, Response] = DurationRecorder(this) { recorded =>
+    val duration = recorded.duration
+    attemptWith(duration)
+    recorded.result.foreach { successfulResponse =>
+      successWith(duration)
+      successfulResponse match {
+        case FullFetch => fullFetchWith(duration)
+        case NotModified => notModifiedWith(duration)
       }
     }
+  }
 
   def keyOn[K2](f: K2 => K): Fetching[K2, Response] = KeyAdapter(this)(f)
 
-  def mapResponse[Response2](f: Response => Response2): Fetching[K, Response2] = ResponseMapper(this)(f)
+  def mapResponse[Response2](f: Response => Response2)(implicit ec: ExecutionContext): Fetching[K, Response2] = ResponseMapper(this)(f)
 
-  def thenParsing[V](parse: Response => V): Loading[K, V] = Loading.by(this)(parse)
+  def thenParsing[V](parse: Response => V)(implicit parsingEC: ExecutionContext): Loading[K, V] = Loading.by(this)(parse)
 }
 
 object Fetching {
@@ -47,38 +49,44 @@ object Fetching {
   object DurationRecorder {
     case class Result(duration: Duration, result: Try[SuccessfulFetch])
   }
-  case class DurationRecorder[K, Response](underlying: Fetching[K, Response])(recorder: Result => Unit)
+
+  /**
+   * @param ec should recording a result happen on the same thread that
+   */
+  case class DurationRecorder[K, Response](underlying: Fetching[K, Response])(recorder: Result => Unit)(implicit ec: ExecutionContext)
     extends Fetching[K, Response] {
 
-    private def time[V](block: => Future[V])(f: V => SuccessfulFetch)(implicit ec: ExecutionContext): Future[V] = {
+    private def time[V](block: => Future[V])(f: V => SuccessfulFetch): Future[V] = {
       val start = Instant.now()
       val resultF = block
       resultF.onComplete(resultTry => recorder(Result(Duration.between(start, Instant.now()), resultTry.map(f))))
       resultF
     }
 
-    override def fetch(key: K)(implicit ec: ExecutionContext): Future[MissingOrETagged[Response]] =
+    override def fetch(key: K): Future[MissingOrETagged[Response]] =
       time(underlying.fetch(key))(_ => FullFetch)
 
-    override def fetchOnlyIfETagChanged(key: K, eTag: String)(implicit ec: ExecutionContext): Future[Option[MissingOrETagged[Response]]] =
+    override def fetchOnlyIfETagChanged(key: K, eTag: String): Future[Option[MissingOrETagged[Response]]] =
       time(underlying.fetchOnlyIfETagChanged(key, eTag))(_.map(_ => FullFetch).getOrElse(NotModified))
   }
 
   private case class KeyAdapter[K, UnderlyingK, Response](underlying: Fetching[UnderlyingK, Response])(f: K => UnderlyingK)
     extends Fetching[K, Response] {
-    override def fetch(key: K)(implicit ec: ExecutionContext): Future[MissingOrETagged[Response]] =
+    override def fetch(key: K): Future[MissingOrETagged[Response]] =
       underlying.fetch(f(key))
 
-    override def fetchOnlyIfETagChanged(key: K, eTag: String)(implicit ec: ExecutionContext): Future[Option[MissingOrETagged[Response]]] =
+    override def fetchOnlyIfETagChanged(key: K, eTag: String): Future[Option[MissingOrETagged[Response]]] =
       underlying.fetchOnlyIfETagChanged(f(key), eTag)
   }
 
-  private case class ResponseMapper[K, UnderlyingResponse, Response](underlying: Fetching[K, UnderlyingResponse])(f: UnderlyingResponse => Response)
+  private case class ResponseMapper[K, UnderlyingResponse, Response](underlying: Fetching[K, UnderlyingResponse])(
+    f: UnderlyingResponse => Response
+  )(implicit responseMappingEC: ExecutionContext) // the function `f` _may_ be resource intensive, and require an appropriate ExecutionContext
     extends Fetching[K, Response] {
-    override def fetch(key: K)(implicit ec: ExecutionContext): Future[MissingOrETagged[Response]] =
+    override def fetch(key: K): Future[MissingOrETagged[Response]] =
       underlying.fetch(key).map(_.map(f))
 
-    override def fetchOnlyIfETagChanged(key: K, eTag: String)(implicit ec: ExecutionContext): Future[Option[MissingOrETagged[Response]]] =
+    override def fetchOnlyIfETagChanged(key: K, eTag: String): Future[Option[MissingOrETagged[Response]]] =
       underlying.fetchOnlyIfETagChanged(key, eTag).map(_.map(_.map(f)))
   }
 }

core/src/test/scala/com/gu/etagcaching/FreshnessPolicyTest.scala

@@ -63,10 +63,10 @@ class FreshnessPolicyTest extends AnyFlatSpec with Matchers with ScalaFutures wi
     val fetching: Fetching[String, Int] = TestFetching.withIncrementingValues
 
     val eTagCache = new ETagCache[String, Int](
-      fetching.thenParsing(identity),
+      fetching.thenParsing(identity)(ExecutionContext.parasitic),
       TolerateOldValueWhileRefreshing,
       _.maximumSize(1).expireAfterWrite(100.millis)
-    )(ExecutionContext.Implicits.global)
+    )
 
     eTagCache.get("KEY").futureValue.value shouldBe 0
     eTagCache.get("KEY").futureValue.value shouldBe 0

core/src/test/scala/com/gu/etagcaching/TestFetching.scala

@@ -10,11 +10,11 @@ object TestFetching {
   def withIncrementingValues: Fetching[String, Int] = new Fetching[String, Int] {
     val counter = new AtomicInteger()
 
-    override def fetch(key: String)(implicit ec: ExecutionContext): Future[MissingOrETagged[Int]] = {
+    override def fetch(key: String): Future[MissingOrETagged[Int]] = {
       val count = counter.getAndIncrement()
       Future.successful(ETaggedData(count.toString, count))
     }
-    override def fetchOnlyIfETagChanged(key: String, eTag: String)(implicit ec: ExecutionContext): Future[Option[MissingOrETagged[Int]]] =
-      fetch(key)(ec).map(Some(_))
+    override def fetchOnlyIfETagChanged(key: String, eTag: String): Future[Option[MissingOrETagged[Int]]] =
+      fetch(key).map(Some(_))(ExecutionContext.parasitic)
   }
 }