LIVE-6468 Reuse http client across invocations

notificationworkerlambda/cdk/lib/senderworker.ts

@@ -107,7 +107,7 @@ class SenderWorker extends cdkcore.Construct  {
         Stage: scope.stage,
         Stack: scope.stack,
         App: id,
-        Platform: props.platform
+        Platform: props.platform,
       },
       memorySize: 10240,
       description: `sends notifications for ${id}`,

notificationworkerlambda/src/main/scala/com/gu/notifications/worker/AndroidSender.scala

@@ -8,7 +8,7 @@ import com.gu.notifications.worker.delivery.DeliveryException.InvalidToken
 import com.gu.notifications.worker.delivery.{BatchDeliverySuccess, DeliveryClient, DeliveryException}
 import com.gu.notifications.worker.delivery.fcm.{Fcm, FcmClient}
 import com.gu.notifications.worker.models.{LatencyMetrics, SendingResults}
-import com.gu.notifications.worker.tokens.{BatchNotification, ChunkedTokens}
+import com.gu.notifications.worker.tokens.{BatchNotification, ChunkedTokens, IndividualNotification}
 import com.gu.notifications.worker.utils.{Cloudwatch, CloudwatchImpl, Reporting}
 import fs2.{Pipe, Stream}
 
@@ -40,18 +40,24 @@ class AndroidSender(val config: FcmWorkerConfiguration, val firebaseAppName: Opt
   override implicit val timer: Timer[IO] = IO.timer(ec)
 
   val fcmFirebase: Try[FcmFirebase] = FcmFirebase(config.fcmConfig, firebaseAppName)
+
+  val fcmClients: Seq[Try[FcmClient]] = Seq.fill(20)(fcmFirebase).map(firebase => firebase.map(FcmClient(_)))
+
+  val deliveryServiceStream: Stream[IO, Fcm[IO]] = 
+    Stream.emits(fcmClients).covary[IO].flatMap(_.fold(e => Stream.raiseError[IO](e), c => Stream.eval[IO, Fcm[IO]]( IO.delay(new Fcm(c)))))
+
   override val deliveryService: IO[Fcm[IO]] = 
-    fcmFirebase.fold(e => IO.raiseError(e), c => IO.delay(new Fcm(FcmClient(c))))
+    fcmFirebase.map(FcmClient(_)).fold(e => IO.raiseError(e), c => IO.delay(new Fcm(c)))
 
   override val maxConcurrency = config.concurrencyForIndividualSend
   override val batchConcurrency = 100
 
   //override the deliverChunkedTokens method to validate the success of sending batch notifications to the FCM client. This implementation could be refactored in the future to make it more streamlined with APNs
   override def deliverChunkedTokens(chunkedTokenStream: Stream[IO, (ChunkedTokens, Long, Instant, Int)]): Stream[IO, Unit] = {
-    chunkedTokenStream.map {
-      case (chunkedTokens, sentTime, functionStartTime, sqsMessageBatchSize) =>
+    chunkedTokenStream.parZip(deliveryServiceStream.repeat).map {
+      case ((chunkedTokens, sentTime, functionStartTime, sqsMessageBatchSize), deliveryServiceForInd) =>
         if (config.isIndividualSend(chunkedTokens.notification.topic.map(_.toString()))) 
-          deliverIndividualNotificationStream(Stream.emits(chunkedTokens.toNotificationToSends).covary[IO])
+          deliverIndividualNotificationStream(Stream.emits(chunkedTokens.toNotificationToSends).covary[IO], deliveryServiceForInd)
                       .broadcastTo(
                         reportSuccesses(chunkedTokens, sentTime, functionStartTime, sqsMessageBatchSize),
                         cleanupFailures,
@@ -68,6 +74,12 @@ class AndroidSender(val config: FcmWorkerConfiguration, val firebaseAppName: Opt
     }.parJoin(batchConcurrency)
   }
 
+  def deliverIndividualNotificationStream(individualNotificationStream: Stream[IO, IndividualNotification], deliveryService: Fcm[IO]): Stream[IO, Either[DeliveryException, FcmClient#Success]] = for {
+    resp <- individualNotificationStream.map(individualNotification => deliveryService.send(individualNotification.notification, individualNotification.token))
+      .parJoin(maxConcurrency)
+      .evalTap(Reporting.log(s"Sending failure: "))
+  } yield resp
+
   def deliverBatchNotificationStream[C <: FcmClient](batchNotificationStream: Stream[IO, BatchNotification]): Stream[IO, Either[DeliveryException, C#BatchSuccess]] = for {
     deliveryService <- Stream.eval(deliveryService)
     resp <- batchNotificationStream.map(batchNotification => deliveryService.sendBatch(batchNotification.notification, batchNotification.token))
@@ -83,7 +95,7 @@ class AndroidSender(val config: FcmWorkerConfiguration, val firebaseAppName: Opt
     }
     input
       .fold(SendingResults.empty) { case (acc, resp) => SendingResults.aggregateBatch(acc, chunkedTokens.tokens.size, resp) }
-      .evalTap(logInfoWithFields(logFields(env, chunkedTokens.notification, chunkedTokens.tokens.size, sentTime, functionStartTime, Configuration.platform, sqsMessageBatchSize), prefix = s"Results $notificationLog: ").andThen(_.map(cloudwatch.sendPerformanceMetrics(env.stage, enableAwsMetric))))
+      .evalTap(logInfoWithFields(logFields(env, chunkedTokens.notification, chunkedTokens.tokens.size, sentTime, functionStartTime, Configuration.platform, sqsMessageBatchSize, messagingApi = Some("Batch")), prefix = s"Results $notificationLog: ").andThen(_.map(cloudwatch.sendPerformanceMetrics(env.stage, enableAwsMetric))))
       .through(cloudwatch.sendResults(env.stage, Configuration.platform))
   }
 

notificationworkerlambda/src/main/scala/com/gu/notifications/worker/SenderRequestHandler.scala

@@ -43,7 +43,7 @@ trait SenderRequestHandler[C <: DeliveryClient] extends Logging {
     }
 
     input.fold(SendingResults.empty) { case (acc, resp) => SendingResults.aggregate(acc, resp) }
-      .evalTap(logInfoWithFields(logFields(env, chunkedTokens.notification, chunkedTokens.tokens.size, sentTime, functionStartTime, Configuration.platform, sqsMessageBatchSize = sqsMessageBatchSize), prefix = s"Results $notificationLog: ").andThen(_.map(cloudwatch.sendPerformanceMetrics(env.stage, enableAwsMetric))))
+      .evalTap(logInfoWithFields(logFields(env, chunkedTokens.notification, chunkedTokens.tokens.size, sentTime, functionStartTime, Configuration.platform, sqsMessageBatchSize = sqsMessageBatchSize, messagingApi = Some("Individual")), prefix = s"Results $notificationLog: ").andThen(_.map(cloudwatch.sendPerformanceMetrics(env.stage, enableAwsMetric))))
       .through(cloudwatch.sendResults(env.stage, Configuration.platform))
   }
 

notificationworkerlambda/src/main/scala/com/gu/notifications/worker/delivery/fcm/FcmClient.scala

@@ -76,7 +76,7 @@ class FcmClient (firebaseMessaging: FirebaseMessaging, firebaseApp: FirebaseApp,
           logger.info(Map(
             "worker.individualRequestLatency" -> Duration.between(start, requestCompletionTime).toMillis,
             "notificationId" -> notificationId,
-          ), "Individual send request completed")
+          ), s"Individual send request completed - ${this.toString()}")
           onAPICallComplete(parseSendResponse(notificationId, token, response, requestCompletionTime))
         }
     }

notificationworkerlambda/src/main/scala/com/gu/notifications/worker/utils/Logging.scala

@@ -51,6 +51,7 @@ trait Logging {
     functionStartTime: Instant,
     maybePlatform: Option[Platform],
     sqsMessageBatchSize: Int,
+    messagingApi: Option[String] = None
   )(end: Instant): Map[String, Any] = {
     val processingTime = Duration.between(functionStartTime, end).toMillis
     val processingRate = numberOfTokens.toDouble / processingTime * 1000
@@ -67,7 +68,7 @@ trait Logging {
       "worker.notificationProcessingEndTime.millis" -> end.toEpochMilli,
       "sqsMessageBatchSize" -> sqsMessageBatchSize,
       "worker.chunkTokenSize" -> numberOfTokens,
-    )
+    ) ++ messagingApi.map(s => Map("worker.messagingApi" -> s)).getOrElse(Map())
   }
 
   def logStartAndCount(acc: Int, chunkedTokens: ChunkedTokens): Int = {