Jak se v Slicku používají reaktivní streamy pro vkládání dat

Sériové přílohy

Nejjednodušší způsob by bylo udělat vložení v rámci Sink.foreach .

Za předpokladu, že jste použili generování kódu schématu a dále za předpokladu, že se vaše tabulka jmenuje "NumberTable"

//Tables file was auto-generated by the schema code generation
import Tables.{Numbertable, NumbertableRow} 

val numberTableDB = Database forConfig "NumberTableConfig"

Můžeme napsat funkci, která provede vkládání

def insertIntoDb(num : Int) = 
  numberTableDB run (Numbertable += NumbertableRow(num))

A tato funkce může být umístěna do Sink

val insertSink = Sink[Int] foreach insertIntoDb

Source(0 to 100) runWith insertSink

Dávkové přílohy

Metodiku Sink můžete dále rozšířit dávkováním N vložek najednou:

def batchInsertIntoDb(nums : Seq[Int]) = 
  numberTableDB run (Numbertable ++= nums.map(NumbertableRow.apply))

val batchInsertSink = Sink[Seq[Int]] foreach batchInsertIntoDb

Tento dávkový dřez může být napájen Flow který dělá dávkové seskupení:

val batchSize = 10

Source(0 to 100).via(Flow[Int].grouped(batchSize))
                .runWith(batchInsertSink)