def atLeastOnceAsync[Offset, Envelope]()

in projection/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcProjection.scala [154:184]

• 28 lines of code
• 1 McCabe index (conditional complexity)

  def atLeastOnceAsync[Offset, Envelope](
      projectionId: ProjectionId,
      settings: Option[R2dbcProjectionSettings],
      sourceProvider: SourceProvider[Offset, Envelope],
      handler: () => Handler[Envelope])(implicit system: ActorSystem[_]): AtLeastOnceProjection[Offset, Envelope] = {

    val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(system))
    val connFactory = connectionFactory(system, r2dbcSettings)
    val offsetStore =
      R2dbcProjectionImpl.createOffsetStore(
        projectionId,
        timestampOffsetBySlicesSourceProvider(sourceProvider),
        r2dbcSettings,
        connFactory)

    val adaptedHandler =
      R2dbcProjectionImpl.adaptedHandlerForAtLeastOnceAsync(sourceProvider, handler, offsetStore)(
        system.executionContext,
        system)

    new R2dbcProjectionImpl(
      projectionId,
      r2dbcSettings,
      settingsOpt = None,
      sourceProvider,
      restartBackoffOpt = None,
      offsetStrategy = AtLeastOnce(),
      handlerStrategy = SingleHandlerStrategy(adaptedHandler),
      NoopStatusObserver,
      offsetStore)
  }