<commit_before>package hmda.publication.lar

import akka.actor.typed.Behavior
import akka.actor.typed.scaladsl._
import akka.actor.typed.scaladsl.adapter._
import akka.actor.{ ActorSystem => UntypedActorSystem }
import akka.kafka.CommitterSettings
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Keep
import com.amazonaws.services.simpleemail.AmazonSimpleEmailServiceClientBuilder
import hmda.publication.lar.config.Settings
import hmda.publication.lar.database.{ EmailSubmissionStatusRepository, PGSlickEmailSubmissionStatusRepository }
import hmda.publication.lar.email.SESEmailService
import hmda.publication.lar.streams.Stream.{ commitMessages, pullEmails, sendEmailsIfNecessary }
import monix.execution.Scheduler
import slick.basic.DatabaseConfig
import slick.jdbc.JdbcProfile

object EmailGuardian {
  sealed trait GuardianProtocol
  case object Ready                 extends GuardianProtocol
  case class Error(message: String) extends GuardianProtocol

  def behavior: Behavior[GuardianProtocol] =
    Behaviors.setup { ctx =>
      implicit val system: UntypedActorSystem = ctx.system.toUntyped
      implicit val mat: ActorMaterializer     = ActorMaterializer()
      implicit val monixScheduler: Scheduler  = Scheduler(ctx.executionContext)

      val databaseConfig                                   = DatabaseConfig.forConfig[JdbcProfile]("db")
      val config                                           = Settings(system)
      val serviceClient                                    = AmazonSimpleEmailServiceClientBuilder.defaultClient()
      val emailService                                     = new SESEmailService(serviceClient, config.email.fromAddress)
      val emailStatusRepo: EmailSubmissionStatusRepository = new PGSlickEmailSubmissionStatusRepository(databaseConfig)
      val commitSettings                                   = CommitterSettings(config.kafka.commitSettings)

      val (control, streamCompletion) =
        pullEmails(system, config.kafka.bootstrapServers)
          .via(sendEmailsIfNecessary(emailService, emailStatusRepo, config.email.content, config.email.subject, config.email.parallelism,4,config.bankFilterList))
          .asSource
          .map { case (_, offset) => offset }
          .toMat(commitMessages(commitSettings))(Keep.both)
          .run()

      streamCompletion.onComplete { t =>
        val errorMessage = t.fold(e => e.getMessage, _ => "infinite stream completed")
        ctx.self ! Error(errorMessage)
      }

      Behaviors.receiveMessage {
        case Error(errorMessage) =>
          ctx.log.error(s"Infinite consumer stream has terminated, Error: $errorMessage")
          control.drainAndShutdown(streamCompletion).onComplete { _ =>
            serviceClient.shutdown()
            databaseConfig.db.shutdown
          }
          Behaviors.stopped
      }
    }
}<commit_msg> removing hard coded reference to time to retry<commit_after>package hmda.publication.lar

import akka.actor.typed.Behavior
import akka.actor.typed.scaladsl._
import akka.actor.typed.scaladsl.adapter._
import akka.actor.{ ActorSystem => UntypedActorSystem }
import akka.kafka.CommitterSettings
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Keep
import com.amazonaws.services.simpleemail.AmazonSimpleEmailServiceClientBuilder
import hmda.publication.lar.config.Settings
import hmda.publication.lar.database.{ EmailSubmissionStatusRepository, PGSlickEmailSubmissionStatusRepository }
import hmda.publication.lar.email.SESEmailService
import hmda.publication.lar.streams.Stream.{ commitMessages, pullEmails, sendEmailsIfNecessary }
import monix.execution.Scheduler
import slick.basic.DatabaseConfig
import slick.jdbc.JdbcProfile

object EmailGuardian {
  sealed trait GuardianProtocol
  case object Ready                 extends GuardianProtocol
  case class Error(message: String) extends GuardianProtocol

  def behavior: Behavior[GuardianProtocol] =
    Behaviors.setup { ctx =>
      implicit val system: UntypedActorSystem = ctx.system.toUntyped
      implicit val mat: ActorMaterializer     = ActorMaterializer()
      implicit val monixScheduler: Scheduler  = Scheduler(ctx.executionContext)

      val databaseConfig                                   = DatabaseConfig.forConfig[JdbcProfile]("db")
      val config                                           = Settings(system)
      val serviceClient                                    = AmazonSimpleEmailServiceClientBuilder.defaultClient()
      val emailService                                     = new SESEmailService(serviceClient, config.email.fromAddress)
      val emailStatusRepo: EmailSubmissionStatusRepository = new PGSlickEmailSubmissionStatusRepository(databaseConfig)
      val commitSettings                                   = CommitterSettings(config.kafka.commitSettings)

      val (control, streamCompletion) =
        pullEmails(system, config.kafka.bootstrapServers)
          .via(sendEmailsIfNecessary(emailService, emailStatusRepo, config.email.content, config.email.subject, config.email.parallelism,config.email.timeToRetry,config.bankFilterList))
          .asSource
          .map { case (_, offset) => offset }
          .toMat(commitMessages(commitSettings))(Keep.both)
          .run()

      streamCompletion.onComplete { t =>
        val errorMessage = t.fold(e => e.getMessage, _ => "infinite stream completed")
        ctx.self ! Error(errorMessage)
      }

      Behaviors.receiveMessage {
        case Error(errorMessage) =>
          ctx.log.error(s"Infinite consumer stream has terminated, Error: $errorMessage")
          control.drainAndShutdown(streamCompletion).onComplete { _ =>
            serviceClient.shutdown()
            databaseConfig.db.shutdown
          }
          Behaviors.stopped
      }
    }
}
{'fraction_non_alphanumeric': 0.07582285024987076, 'fraction_numerical': 0.00017232465965879716, 'mean_word_length': 3.0, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2277500', 'n_tokens_mistral': 1539, 'n_tokens_neox': 1493, 'n_words': 286}