Working Channel API and Broadcast API clients

Author: waisingyiu

liveactivities/src/main/scala/com/gu/liveactivities/BroadcastApiClient.scala

@@ -56,7 +56,15 @@ class BroadcastApiClient {
   private val startPayload: String = Json.stringify(Json.toJson(broadcastStartBodyFixture))
 
   private def getAccessToken(): String = {
-    return "invalid-token-for-testing"
+    ChannelApiClient.authenticationToken match {
+      case Some(token) => token
+      case None => {
+        val authenticationToken = generateToken()
+        ChannelApiClient.authenticationToken = Some(authenticationToken)
+        ChannelApiClient.issueDate = Some(new Date())
+        authenticationToken
+      }
+    }
   }
 
   private def generateToken(): String = {
@@ -71,19 +79,17 @@ class BroadcastApiClient {
 
   def sendToChannel(channelId: String, expiration: Option[Instant], priority: Option[Int]): Future[String] = {
     println(s"Broadcasting to channel $channelId")
-    val authToken = generateToken()
+    val authToken = getAccessToken()
     println(s"Generated auth token: $authToken")
 
     val request: HttpRequest = HttpRequest.newBuilder(new URI(url))
       .version(HttpClient.Version.HTTP_2)
       .header("Authorization", authToken)
       .header("Content-Type", mediaType)
+      .header("apns-channel-id", channelId)
       .header("apns-expiration", expiration.getOrElse(Instant.now().plusSeconds(5 * 60)).getEpochSecond.toString)
       .header("apns-priority", priority.map(_.toString).getOrElse("1"))
       .header("apns-push-type", "Liveactivity")
-
-      // add channel id
-      .header("apns-channel-id", channelId)
       .POST(HttpRequest.BodyPublishers.ofString(startPayload, charSet))
 
       .timeout(Duration.ofSeconds(60))

liveactivities/src/main/scala/com/gu/liveactivities/ChannelApiClient.scala

@@ -13,6 +13,14 @@ import com.turo.pushy.apns.auth.ApnsSigningKey;
 import com.turo.pushy.apns.auth.AuthenticationToken;
 import java.time.Instant
 import java.util.Date
+import com.gu.liveactivities.ChannelApiClient.authenticationToken
+
+object ChannelApiClient {
+
+  var authenticationToken: Option[String] = None
+
+  var issueDate: Option[Date] = None
+}
 
 class ChannelApiClient {
 
@@ -31,7 +39,15 @@ class ChannelApiClient {
   private val message = "{\"message-storage-policy\": 1, \"push-type\": \"LiveActivity\"}"
 
   private def getAccessToken(): String = {
-    return "invalid-token-for-testing"
+    ChannelApiClient.authenticationToken match {
+      case Some(token) => token
+      case None => {
+        val authenticationToken = generateToken()
+        ChannelApiClient.authenticationToken = Some(authenticationToken)
+        ChannelApiClient.issueDate = Some(new Date())
+        authenticationToken
+      }
+    }
   }
 
   private def generateToken(): String = {
@@ -46,9 +62,7 @@ class ChannelApiClient {
 
   def createChannel(): Future[String] = {
     println("Creating channel")
-    val authToken = generateToken()
-    println(s"Generated auth token: $authToken")
-
+    val authToken = getAccessToken()
     val request: HttpRequest = HttpRequest.newBuilder(new URI(url))
       .version(HttpClient.Version.HTTP_2)
       .header("Authorization", authToken)
@@ -60,11 +74,46 @@ class ChannelApiClient {
     httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString()).whenComplete((response, err) => {
       if (response == null) {
         p.failure(err)
+      } else if (response.statusCode() >= 200 && 
+                response.statusCode() < 300 && 
+                response.headers().firstValue("apns-channel-id").isPresent()) {
+        val channelId = response.headers().firstValue("apns-channel-id").get()
+        println(s"Channel created successfully with channel ID $channelId")
+        p.success(channelId)
       } else {
-        println(s"Received response with status code ${response.statusCode()} and body ${response.body()}")
-        p.success(response.body())
+        println(s"Failed to create channel with status code ${response.statusCode()} and body ${response.body()}")
+        p.failure(new Exception(s"Failed to create channel with status code ${response.statusCode()} and body ${response.body()}"))
       }
     })
     p.future
   }
+
+  def closeChannel(channelId: String): Future[Unit] = {
+    println(s"Closing channel $channelId")
+    val authToken = getAccessToken()
+    val request: HttpRequest = HttpRequest.newBuilder(new URI(url))
+      .version(HttpClient.Version.HTTP_2)
+      .header("Authorization", authToken)
+      .header("Content-Type", mediaType)
+      .header("apns-channel-id", channelId)
+      .DELETE()
+      .timeout(Duration.ofSeconds(60))
+      .build()
+    val p = Promise[Unit]()
+    httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString()).whenComplete((response, err) => {
+      if (response == null) {
+        println(s"Failed to close channel $channelId due to error ${err.getMessage}")
+        p.failure(err)
+      } else if (response.statusCode() >= 200 && 
+                response.statusCode() < 300) {
+        println(s"Channel closed successfully with channel ID $channelId")
+        p.success(())
+      } else {
+        println(s"Failed to close channel with status code ${response.statusCode()} and body ${response.body()}")
+        p.failure(new Exception(s"Failed to close channel with status code ${response.statusCode()} and body ${response.body()}"))
+      }
+    })
+    p.future
+  } 
+
 }

liveactivities/src/main/scala/com/gu/liveactivities/ChannelManagerLambda.scala

@@ -5,15 +5,39 @@ object ChannelManagerLambda {
 
   implicit val ec: scala.concurrent.ExecutionContext = scala.concurrent.ExecutionContext.global
 
+  val channelApiClient = new ChannelApiClient()
+  val broadcastApiClient = new BroadcastApiClient()
+
+  def onChannelCreated(channelId: String): Unit = {
+    println(s"Message sent to channel with ID: $channelId")
+    // You can add additional logic here to handle the sent message, such as logging or sending a notification.
+
+    broadcastApiClient.sendToChannel(channelId, None, None).onComplete {
+      case scala.util.Success(messageId) => {
+        println(s"Message sent successfully with message ID $messageId")
+        onMessageSent(channelId)
+      }
+      case scala.util.Failure(exception) => println(s"Failed to send message to channel $channelId: ${exception.getMessage}")
+    }
+  }
+
+
+  def onMessageSent(channelId: String): Unit = {
+    Thread.sleep(10 * 1000) // Wait for 10 seconds before closing the channel
+    channelApiClient.closeChannel(channelId).onComplete {
+      case scala.util.Success(_) => println(s"Channel $channelId closed successfully")
+      case scala.util.Failure(exception) => println(s"Failed to close channel $channelId: ${exception.getMessage}")
+    }
+  }
+
   def handleRequest(): Unit = {
-    val client = new ChannelApiClient()
-    val channelFuture = client.createChannel()
+    val channelFuture = channelApiClient.createChannel()
     channelFuture.onComplete {
-      case scala.util.Success(channelId) => println(s"Channel created successfully with id: $channelId")
+      case scala.util.Success(channelId) => onChannelCreated(channelId)
       case scala.util.Failure(exception) => println(s"Failed to create channel: ${exception.getMessage}")
     }
      // Keep the main thread alive to allow the async operation to complete
-    Thread.sleep(5000)
+    Thread.sleep(30000) 
   }
 }