blob: 307dd819972011930197b264066be0bc2e49cf6f (
plain) (
blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
package triad
import akka.NotUsed
import akka.stream.scaladsl.{
BroadcastHub,
Keep,
Source,
SourceQueueWithComplete
}
import akka.stream.{Materializer, OverflowStrategy, QueueOfferResult}
import scala.concurrent.Future
/** A very basic streaming message router. */
class LiveMessages(implicit materializer: Materializer) {
private val (
in: SourceQueueWithComplete[Message],
out: Source[Message, NotUsed]
) = Source
.queue[Message](10, OverflowStrategy.dropTail)
.toMat(BroadcastHub.sink[Message])(Keep.both)
.run()
/** Push a single message into the stream. */
def push(message: Message): Future[QueueOfferResult] = in.offer(message)
/** Obtain a stream of the message feed. */
def feed: Source[Message, NotUsed] = out
}
|
