co.paralleluniverse.strands.channels.reactivestreams

Class ReactiveStreams

java.lang.Object

co.paralleluniverse.strands.channels.reactivestreams.ReactiveStreams

public class ReactiveStreams
extends Object

Converts between Quasar channels and reactive streams

Constructor Summary

Constructors

Constructor and Description
ReactiveStreams()

Method Summary

Modifier and Type	Method and Description
static <T> co.paralleluniverse.strands.channels.ReceivePort<T>	subscribe(int bufferSize, co.paralleluniverse.strands.channels.Channels.OverflowPolicy policy, org.reactivestreams.Publisher<T> publisher) Subscribes to a given Publisher and return a ReceivePort to the subscription.
static <T,R> org.reactivestreams.Processor<T,R>	toProcessor(co.paralleluniverse.fibers.FiberFactory ff, int bufferSize, co.paralleluniverse.strands.channels.Channels.OverflowPolicy policy, co.paralleluniverse.strands.SuspendableAction2<? extends co.paralleluniverse.strands.channels.ReceivePort<? super T>,? extends co.paralleluniverse.strands.channels.SendPort<? extends R>> transformer) Turns a transformer into a Publisher.
static <T,R> org.reactivestreams.Processor<T,R>	toProcessor(int bufferSize, co.paralleluniverse.strands.channels.Channels.OverflowPolicy policy, co.paralleluniverse.strands.SuspendableAction2<? extends co.paralleluniverse.strands.channels.ReceivePort<? super T>,? extends co.paralleluniverse.strands.channels.SendPort<? extends R>> transformer) Turns a transformer into a Publisher.
static <T> org.reactivestreams.Publisher<T>	toPublisher(co.paralleluniverse.strands.channels.ReceivePort<T> channel) Turns a channel to a Publisher.
static <T> org.reactivestreams.Publisher<T>	toPublisher(co.paralleluniverse.strands.channels.ReceivePort<T> channel, co.paralleluniverse.fibers.FiberFactory ff) Turns a channel to a Publisher.
static <T> org.reactivestreams.Publisher<T>	toPublisher(co.paralleluniverse.strands.channels.Topic<T> topic) Turns a topic to a Publisher.
static <T> org.reactivestreams.Publisher<T>	toPublisher(co.paralleluniverse.strands.channels.Topic<T> topic, co.paralleluniverse.fibers.FiberFactory ff) Turns a topic to a Publisher.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

ReactiveStreams

public ReactiveStreams()

Method Detail

subscribe

public static <T> co.paralleluniverse.strands.channels.ReceivePort<T> subscribe(int bufferSize,
                                                                                co.paralleluniverse.strands.channels.Channels.OverflowPolicy policy,
                                                                                org.reactivestreams.Publisher<T> publisher)

Subscribes to a given Publisher and return a ReceivePort to the subscription. This creates an internal single consumer channel that will receive the published elements.

Parameters:

bufferSize - the size of the buffer of the internal channel; may be -1 for unbounded, but may not be 0)

policy - the Channels.OverflowPolicy of the internal channel.

publisher - the subscriber

Returns:

A ReceivePort which emits the elements published by the subscriber

toPublisher

public static <T> org.reactivestreams.Publisher<T> toPublisher(co.paralleluniverse.strands.channels.ReceivePort<T> channel,
                                                               co.paralleluniverse.fibers.FiberFactory ff)

Turns a channel to a Publisher. All items sent to the channel will be published by the publisher.

The publisher will allow a single subscription, unless the channel is a ticker channel in which case, multiple subscribers will be allowed, and a new ticker consumer will be created for each.

Every subscription to the returned publisher creates an internal fiber, that will receive items from the channel and publish them.

Parameters:

channel - the channel

ff - the FiberFactory to create the internal fiber(s); if null then a default factory is used.

Returns:

a new publisher for the channel's items

toPublisher

public static <T> org.reactivestreams.Publisher<T> toPublisher(co.paralleluniverse.strands.channels.ReceivePort<T> channel)

Turns a channel to a Publisher. All items sent to the channel will be published by the publisher.

The publisher will allow a single subscription, unless the channel is a ticker channel in which case, multiple subscribers will be allowed, and a new ticker consumer will be created for each.

Every subscription to the returned publisher creates an internal fiber, that will receive items from the channel and publish them.

Calling this method is the same as calling {@link #toPublisher(ReceivePort, FiberFactory) toPublisher(channel, null)

Parameters:

channel - the channel

Returns:

a new publisher for the channel's items

toPublisher

public static <T> org.reactivestreams.Publisher<T> toPublisher(co.paralleluniverse.strands.channels.Topic<T> topic,
                                                               co.paralleluniverse.fibers.FiberFactory ff)

Turns a topic to a Publisher. All items sent to the topic will be published by the publisher.

A new transfer channel (i.e. a blocking channel with a buffer of size 0) subscribed to the topic will be created for every subscriber.

Every subscription to the returned publisher creates an internal fiber, that will receive items from the subscription's channel and publish them.

Parameters:

topic - the topic

ff - the FiberFactory to create the internal fiber(s); if null then a default factory is used.

Returns:

a new publisher for the topic's items

toPublisher

public static <T> org.reactivestreams.Publisher<T> toPublisher(co.paralleluniverse.strands.channels.Topic<T> topic)

Turns a topic to a Publisher. All items sent to the topic will be published by the publisher.

A new transfer channel (i.e. a blocking channel with a buffer of size 0) subscribed to the topic will be created for every subscriber.

Every subscription to the returned publisher creates an internal fiber, that will receive items from the subscription's channel and publish them.

Calling this method is the same as calling {@link #toPublisher(ReceivePort, FiberFactory) toPublisher(channel, null)

Parameters:

topic - the topic

Returns:

a new publisher for the topic's items

toProcessor

public static <T,R> org.reactivestreams.Processor<T,R> toProcessor(co.paralleluniverse.fibers.FiberFactory ff,
                                                                   int bufferSize,
                                                                   co.paralleluniverse.strands.channels.Channels.OverflowPolicy policy,
                                                                   co.paralleluniverse.strands.SuspendableAction2<? extends co.paralleluniverse.strands.channels.ReceivePort<? super T>,? extends co.paralleluniverse.strands.channels.SendPort<? extends R>> transformer)

Turns a transformer into a Publisher. The transformer will run in its own fiber.

Type Parameters:

T - the type of elements flowing into the transformer

R - the type of elements flowing out of the transformer

Parameters:

ff - the FiberFactory to create the internal fiber(s); if null then a default factory is used.

bufferSize - the size of the buffer of the internal channel; may be -1 for unbounded, but may not be 0)

policy - the Channels.OverflowPolicy of the internal channel.

batch - if the channel has a bounded buffer, whether to request further elements from the publisher in batches whenever the channel's buffer is depleted, or after consuming each element.

transformer - a function that reads from it's input channel and writes to its output channel

Returns:

a Processor running the given transformer.

toProcessor

public static <T,R> org.reactivestreams.Processor<T,R> toProcessor(int bufferSize,
                                                                   co.paralleluniverse.strands.channels.Channels.OverflowPolicy policy,
                                                                   co.paralleluniverse.strands.SuspendableAction2<? extends co.paralleluniverse.strands.channels.ReceivePort<? super T>,? extends co.paralleluniverse.strands.channels.SendPort<? extends R>> transformer)

Turns a transformer into a Publisher. The transformer will run in its own fiber.

Same as calling {@link #toProcessor(FiberFactory, int, OverflowPolicy, boolean, SuspendableAction2) toProcessor(null, bufferSize, policy, transformer)

Type Parameters:

T - the type of elements flowing into the transformer

R - the type of elements flowing out of the transformer

Parameters:

ff - the FiberFactory to create the internal fiber(s); if null then a default factory is used.

bufferSize - the size of the buffer of the internal channel; may be -1 for unbounded, but may not be 0)

policy - the Channels.OverflowPolicy of the internal channel.

batch - if the channel has a bounded buffer, whether to request further elements from the publisher in batches whenever the channel's buffer is depleted, or after consuming each element.

transformer - a function that reads from it's input channel and writes to its output channel

Returns:

a Processor running the given transformer.