override def lastPersistenceIdSequenceNumber()

in core/src/main/scala/org/apache/pekko/persistence/jdbc/query/dao/legacy/ByteArrayReadJournalDao.scala [62:79]

• 17 lines of code
• 3 McCabe index (conditional complexity)

  override def lastPersistenceIdSequenceNumber(persistenceId: String): Future[Option[Long]] =
    db.run(queries.lastPersistenceIdSequenceNumberQuery(persistenceId).result)

  override def messages(
      persistenceId: String,
      fromSequenceNr: Long,
      toSequenceNr: Long,
      max: Long): Source[Try[(PersistentRepr, Long)], NotUsed] = {
    Source
      .fromPublisher(
        db.stream(
          queries.messagesQuery((persistenceId, fromSequenceNr, toSequenceNr, correctMaxForH2Driver(max))).result))
      .via(serializer.deserializeFlow)
      .map {
        case Success((repr, _, ordering)) => Success(repr -> ordering)
        case Failure(e)                   => Failure(e)
      }
  }