def sink[T](partitioner: (Int, T) ? Int, startAfterNrOfConsumers: Int, bufferSize: Int = defaultBufferSize): Sink[T, Source[T, NotUsed]] = statefulSink(() ? (info, elem) ? info.consumerIdByIdx(partitioner(info.size, elem)), startAfterNrOfConsumers, bufferSize)
* This `statefulSink` should be used when there is a need to keep mutable state in the partition function, * e.g. for implemening round-robin or sticky session kind of routing. If state is not needed the [[#sink]] can * be more convenient to use. *
* @param partitioner Function that decides where to route an element. It is a factory of a function to * to be able to hold stateful variables that are unique for each materialization. The function * takes two parameters; the first is information about active consumers, including an array of consumer * identifiers and the second is the stream element. The function should return the selected consumer * identifier for the given element. The function will never be called when there are no active consumers, * i.e. there is always at least one element in the array of identifiers. * @param startAfterNrOfConsumers Elements are buffered until this number of consumers have been connected. * This is only used initially when the stage is starting up, i.e. it is not honored when consumers have * been removed (canceled). * @param bufferSize Total number of elements that can be buffered. If this buffer is full, the producer * is backpressured. */ @ApiMayChange def statefulSink[T](partitioner: () ? (ConsumerInfo, T) ? Long, startAfterNrOfConsumers: Int, bufferSize: Int = defaultBufferSize): Sink[T, Source[T, NotUsed]] = Sink.fromGraph(new PartitionHub[T](partitioner, startAfterNrOfConsumers, bufferSize))
//interupted temination
val killAll = KillSwitches.shared("terminator") //fix a producer
val fixedSource = Source.tick(1.second, 1.second, "message") .zipWith(Source(1 to 100))((a, b) => s"$a-$b") //connect to PartitionHub which uses function to select sink
val sourceGraph = fixedSource.via(killAll.flow).toMat(PartitionHub.sink( (size, elem) => math.abs(elem.hashCode) % size, startAfterNrOfConsumers = 2, bufferSize = 256))(Keep.right) //materialize the source
val fromSource = sourceGraph.run() //connect to fixedSource freely
fromSource.runForeach(msg => println("subs1: " + msg)) fromSource.runForeach(msg => println("subs2: " + msg)) scala.io.StdIn.readLine() killAll.shutdown() actorSys.terminate()
//partitioner function
def roundRobin(): (PartitionHub.ConsumerInfo, String) ? Long = { var i = -1L (info, elem) => { i += 1 info.consumerIdByIdx((i % info.size).toInt) } } val roundRobinGraph = fixedSource.via(killAll.flow).toMat(PartitionHub.statefulSink( () => roundRobin(),startAfterNrOfConsumers = 2,bufferSize