Skip to content
No description, website, or topics provided.
Scala
Branch: master
Clone or download
Fetching latest commit…
Cannot retrieve the latest commit at this time.
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
.circleci version updates (#12) Aug 1, 2019
project Update sbt to 1.3.0 (#29) Sep 5, 2019
src support for ZIO 1.0.0-RC12 (#31) Sep 12, 2019
.gitattributes Add license and readme Jun 10, 2019
.gitignore Add license and readme Jun 10, 2019
.scalafmt.conf Update scalafmt-core to 2.0.1 (#23) Aug 19, 2019
LICENSE Add license and readme Jun 10, 2019
README.md Set CI badge (#4) Jun 24, 2019
build.sbt support for ZIO 1.0.0-RC12 (#31) Sep 12, 2019
sbt Setup SBT project Jun 10, 2019

README.md

Interop reactive streams

CircleCI

This library provides an interoperability layer for reactive streams.

Reactive Streams Producer and Subscriber

ZIO integrates with Reactive Streams by providing conversions from zio.stream.Stream to org.reactivestreams.Publisher and from zio.stream.Sink to org.reactivestreams.Subscriber and vice versa. Simply import import zio.interop.reactiveStreams._ to make the conversions available.

Examples

First, let's get a few imports out of the way.

import org.reactivestreams.example.unicast._
import zio._
import zio.interop.reactiveStreams._
import zio.stream._

val runtime = new DefaultRuntime {}

We use the following Publisher and Subscriber for the examples:

val publisher = new RangePublisher(3, 10)
val subscriber = new SyncSubscriber[Int] {
  override protected def whenNext(v: Int): Boolean = {
    print(s"$v, ")
    true
  }
}

Publisher to Stream

A Publisher used as a Stream buffers up to qSize elements. If possible, qSize should be a power of two for best performance. The default is 16.

val streamFromPublisher = publisher.toStream(qSize = 16)
runtime.unsafeRun(
  streamFromPublisher.run(Sink.collectAll[Integer])
)

Subscriber to Sink

When running a Stream to a Subscriber, a side channel is needed for signalling failures. For this reason toSink returns a tuple of Promise and Sink. The Promise must be failed on Stream failure. The type parameter on toSink is the error type of the Stream.

val asSink = subscriber.toSink[Throwable]
val failingStream = Stream.range(3, 13) ++ Stream.fail(new RuntimeException("boom!"))
runtime.unsafeRun(
  asSink.flatMap { case (errorP, sink) =>
    failingStream.run(sink).catchAll(errorP.fail)
  }
)

Stream to Publisher

val stream = Stream.range(3, 13)
runtime.unsafeRun(
  stream.toPublisher.flatMap { publisher =>
    UIO(publisher.subscribe(subscriber))
  }
)

Sink to Subscriber

toSubscriber returns a Subscriber and an IO which completes with the result of running the Sink or the error if the Publisher fails. A Sink used as a Subscriber buffers up to qSize elements. If possible, qSize should be a power of two for best performance. The default is 16.

val sink = Sink.collectAll[Integer]
runtime.unsafeRun(
  sink.toSubscriber(qSize = 16).flatMap { case (subscriber, result) => 
    UIO(publisher.subscribe(subscriber)) *> result
  }
)
You can’t perform that action at this time.