Можно ли создать бесконечный поток из таблицы базы данных с помощью Akka Stream

В общем, база данных SQL является «пассивной» конструкцией и не вносит активно изменений, подобных тому, что вы описали. Вы можете только «имитировать» «push» с периодическим опросом, например:

val newRecStream = Source

  // Query for table changes
  .unfold(initState) { lastState =>
    // query for new data since lastState and save the current state into newState...
    Some((newState, newRecords))
  }

  // Throttle to limit the poll frequency
  .throttle(...)  

  // breaks down into individual records...
  .flatMapConcat { newRecords =>
    Source.unfold(newRecords) { pendingRecords =>
      if (records is empty) {
        None
      } else {
        // take one record from pendingRecords and save to newRec.  Save the rest into remainingRecords.
        Some(remainingRecords, newRec)
      }
    }
  }

Обновлено: 24 февраля 2016 г.

Пример псевдокода, основанный на обновлении вопроса от 23.02.2016:

implicit val system = ActorSystem("Publisher")
implicit val materializer = ActorMaterializer()
val db = Database.forConfig("pg-postgres")
val queryLimit = 10
try {
  val completion = Source
    .unfoldAsync(0) { lastRowId =>
      val q = TableQuery[Records].filter(row => row.id > lastRowId).take(queryLimit)
      db.run(q.result).map { recs =>
        Some(recs.last.id, recs)
      }
    }
    .throttle(1, 1.second, 1, ThrottleMode.shaping)
    .flatMapConcat { recs =>
      Source.fromIterator(() => recs.iterator)
    }
    .runForeach { rec =>
      println(s"${rec.id}, ${rec.value}")
    }

  // Block forever
  Await.ready(completion, Duration.Inf)

} catch {
  case ex: Throwable => println(ex)
} finally {
  system.shutdown
  db.close
}

Он будет многократно выполнять запрос в unfoldAsync к базе данных, извлекая не более 10 (queryLimit) записей за раз и отправляя записи вниз по течению (-> throttle -> flatMapConcat -> runForeach). Await в конце фактически заблокирует навсегда.

Обновлено: 25 февраля 2016 г.

Исполняемый код «доказательства концепции»:

import akka.actor.ActorSystem
import akka.stream.{ThrottleMode, ActorMaterializer}
import akka.stream.scaladsl.Source
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}
import scala.concurrent.duration._

object Infinite extends App{
  implicit val system = ActorSystem("Publisher")
  implicit val ec = system.dispatcher
  implicit val materializer = ActorMaterializer()
  case class Record(id: Int, value: String)
  try {
    val completion = Source
      .unfoldAsync(0) { lastRowId =>
        Future {
          val recs = (lastRowId to lastRowId + 10).map(i => Record(i, s"rec#$i"))
          Some(recs.last.id, recs)
        }
      }
      .throttle(1, 1.second, 1, ThrottleMode.Shaping)
      .flatMapConcat { recs =>
        Source.fromIterator(() => recs.iterator)
      }
      .runForeach { rec =>
        println(rec)
      }

    Await.ready(completion, Duration.Inf)

  } catch {
    case ex: Throwable => println(ex)
  } finally {
    system.shutdown
  }
}

Вот рабочий код бесконечной потоковой передачи базы данных. Это было протестировано с миллионами записей, вставленных в базу данных postgresql во время работы потокового приложения -

package infinite.streams.db

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.alpakka.slick.scaladsl.SlickSession
import akka.stream.scaladsl.{Flow, Sink, Source}
import akka.stream.{ActorMaterializer, ThrottleMode}
import org.slf4j.LoggerFactory
import slick.basic.DatabaseConfig
import slick.jdbc.JdbcProfile
import scala.concurrent.duration._
import scala.concurrent.{Await, ExecutionContextExecutor}

case class Record(id: Int, value: String) {
  val content = s"<ROW><ID>$id</ID><VALUE>$value</VALUE></ROW>"
}

object InfiniteStreamingApp extends App {

  println("Starting app...")

  implicit val system: ActorSystem = ActorSystem("Publisher")
  implicit val ec: ExecutionContextExecutor = system.dispatcher
  implicit val materializer: ActorMaterializer = ActorMaterializer()

  println("Initializing database configuration...")
  val databaseConfig: DatabaseConfig[JdbcProfile] = DatabaseConfig.forConfig[JdbcProfile]("postgres3")
  implicit val session: SlickSession = SlickSession.forConfig(databaseConfig)

  import databaseConfig.profile.api._

  class Records(tag: Tag) extends Table[Record](tag, "test2") {
    def id = column[Int]("c1")

    def value = column[String]("c2")

    def * = (id, value) <> (Record.tupled, Record.unapply)
  }

  val db = databaseConfig.db

  println("Prime for streaming...")

  val logic: Flow[(Int, String), (Int, String), NotUsed] = Flow[(Int, String)].map {
    case (id, value) => (id, value.toUpperCase)
  }

  val fetchSize = 5
  try {
    val done = Source
      .unfoldAsync(0) {
      lastId =>
        println(s"Fetching next: $fetchSize records with id > $lastId")
        val query = TableQuery[Records].filter(_.id > lastId).take(fetchSize)
        db.run(query.result.withPinnedSession)
          .map {
            recs => Some(recs.last.id, recs)
          }
    }
      .throttle(5, 1.second, 1, ThrottleMode.shaping)
      .flatMapConcat {
        recs => Source.fromIterator(() => recs.iterator)
      }
      .map(x => (x.id, x.content))
      .via(logic)
      .log("*******Post Transformation******")
//      .runWith(Sink.foreach(r => println("SINK: " + r._2)))
// Use runForeach or runWith(Sink)
      .runForeach(rec => println("REC: " + rec))

    println("Waiting for result....")
    Await.ready(done, Duration.Inf)

  } catch {
    case ex: Throwable => println(ex.getMessage)
  } finally {
    println("Streaming end successfully")
    db.close()
    system.terminate()
  }

}



application.conf

akka {
  loggers = ["akka.event.slf4j.Slf4jLogger"]
  loglevel = "INFO"
}
# Load using SlickSession.forConfig("slick-postgres")
postgres3 {
  profile = "slick.jdbc.PostgresProfile$"
  db {
    dataSourceClass = "slick.jdbc.DriverDataSource"
    properties = {
      driver = "org.postgresql.Driver"
      url = "jdbc:postgresql://localhost/testdb"
      user = "postgres"
      password = "postgres"
    }
    numThreads = 2
  }
}