cloudwatchStream.scala

  private def cloudwatchLogs(logGroup: String, streamName: String): Source[OutputLogEvent, NotUsed] = {

    implicit val ec: ExecutionContextExecutor = ExecutionContext.global

    val credentialsProvider = new AWSStaticCredentialsProvider(new ProfileCredentialsProvider("10pines").getCredentials)

    val logsClient = AWSLogsClientBuilder.standard
        .withCredentials(credentialsProvider)
        .withRegion(Regions.US_WEST_2)
        .withClientConfiguration(new ClientConfiguration())
        .build

    Source.unfoldAsync("") { token =>

      val request = new GetLogEventsRequest(logGroup, streamName)
          .withStartTime(0L)
          .withEndTime(System.currentTimeMillis())

      if (token.nonEmpty) request.setNextToken(token)

      Future { logsClient.getLogEvents(request) }
          .map { logs =>
            val newToken = logs.getNextForwardToken
            if (token == newToken) {
              None
            } else {
              import scala.jdk.CollectionConverters._
              val events: List[OutputLogEvent] = logs.getEvents.asScala.toList
              Some((newToken, events))
            }
          }
    }.mapConcat(identity)
  }