def atLeastOnce[Offset, Envelope, S <: JdbcSession]()

in jdbc/src/main/scala/org/apache/pekko/projection/jdbc/javadsl/JdbcProjection.scala [99:125]

24 lines of code
1 McCabe index (conditional complexity)


  def atLeastOnce[Offset, Envelope, S <: JdbcSession](
      projectionId: ProjectionId,
      sourceProvider: SourceProvider[Offset, Envelope],
      sessionCreator: Supplier[S],
      handler: Supplier[JdbcHandler[Envelope, S]],
      system: ActorSystem[_]): AtLeastOnceProjection[Offset, Envelope] = {

    val sessionFactory = () => sessionCreator.get()
    val offsetStore = JdbcProjectionImpl.createOffsetStore(sessionFactory)(system)

    val adaptedHandler =
      JdbcProjectionImpl.adaptedHandlerForAtLeastOnce(
        sessionFactory,
        () => new JdbcHandlerAdapter(handler.get()),
        offsetStore)

    new JdbcProjectionImpl(
      projectionId,
      new SourceProviderAdapter(sourceProvider),
      sessionFactory = sessionFactory,
      settingsOpt = None,
      restartBackoffOpt = None,
      AtLeastOnce(),
      SingleHandlerStrategy(adaptedHandler),
      NoopStatusObserver,
      offsetStore)
  }