Package co.paralleluniverse.strands.channels

Class TransformingReceivePort<T>

java.lang.Object

co.paralleluniverse.strands.channels.DelegatingReceivePort<T>

co.paralleluniverse.strands.channels.TransformingReceivePort<T>

All Implemented Interfaces:

co.paralleluniverse.common.util.DelegatingEquals, Port<T>, PortAutoCloseable, ReceivePort<T>, java.lang.AutoCloseable

public class TransformingReceivePort<T>
extends DelegatingReceivePort<T>

A ReceivePort with additional functional-transform operations, usually wrapping a plain ReceivePort.

Nested Class Summary

Nested classes/interfaces inherited from interface co.paralleluniverse.strands.channels.ReceivePort

ReceivePort.EOFException

Field Summary

Fields inherited from class co.paralleluniverse.strands.channels.DelegatingReceivePort

target

Method Summary

Modifier and Type	Method	Description
<U> TransformingReceivePort<U>	fiberTransform(FiberFactory fiberFactory, SuspendableAction2<? extends ReceivePort<? super T>,? extends SendPort<? extends U>> transformer, Channel<U> out)	Spawns a fiber that transforms values read from this channel and writes values to the out channel.
<U> TransformingReceivePort<U>	fiberTransform(SuspendableAction2<? extends ReceivePort<? super T>,? extends SendPort<? extends U>> transformer, Channel<U> out)	Spawns a fiber that transforms values read from this channel and writes values to the out channel.
TransformingReceivePort<T>	filter(com.google.common.base.Predicate<T> pred)	Returns a TransformingReceivePort that filters messages that satisfy a predicate from this given channel.
<U> TransformingReceivePort<U>	flatMap(com.google.common.base.Function<T,ReceivePort<U>> f)	Returns a TransformingReceivePort that receives messages that are transformed by a given flat-mapping function from this channel.
void	forEach(SuspendableAction1<T> action)	Performs the given action on each message received by this channel.
<U> TransformingReceivePort<U>	map(com.google.common.base.Function<T,U> f)	Returns a TransformingReceivePort that receives messages that are transformed by a given mapping function from this channel.
<U> TransformingReceivePort<U>	reduce(co.paralleluniverse.common.util.Function2<U,T,U> f, U init)	Returns a TransformingReceivePort from which receiving messages that are transformed from a given channel by a given reduction function.
TransformingReceivePort<T>	take(long count)	Returns a TakeReceivePort that can provide at most count messages from the underlying channel.

Methods inherited from class co.paralleluniverse.strands.channels.DelegatingReceivePort

close, equals, hashCode, isClosed, receive, receive, receive, toString, tryReceive

Methods inherited from class java.lang.Object

clone, finalize, getClass, notify, notifyAll, wait, wait, wait

Method Detail

filter

public TransformingReceivePort<T> filter(com.google.common.base.Predicate<T> pred)

Returns a TransformingReceivePort that filters messages that satisfy a predicate from this given channel. All messages (even those not satisfying the predicate) will be consumed from the original channel; those that don't satisfy the predicate will be silently discarded.

The returned TransformingReceivePort has the same hashCode as channel and is equal to it.

Parameters:

pred - the filtering predicate

Returns:

A TransformingReceivePort that will receive all those messages from the original channel which satisfy the predicate (i.e. the predicate returns true).

map

public <U> TransformingReceivePort<U> map(com.google.common.base.Function<T,U> f)

Returns a TransformingReceivePort that receives messages that are transformed by a given mapping function from this channel.

The returned TransformingReceivePort has the same hashCode as channel and is equal to it.

Parameters:

f - the mapping function

Returns:

a TransformingReceivePort that returns messages that are the result of applying the mapping function to the messages received on the given channel.

reduce

public <U> TransformingReceivePort<U> reduce(co.paralleluniverse.common.util.Function2<U,T,U> f,
                                             U init)

Returns a TransformingReceivePort from which receiving messages that are transformed from a given channel by a given reduction function.

The returned TransformingReceivePort has the same hashCode as channel and is equal to it.

Parameters:

f - The reduction function.

init - The initial input to the reduction function.

Returns:

a ReceivePort that returns messages that are the result of applying the reduction function to the messages received on the given channel.

flatMap

public <U> TransformingReceivePort<U> flatMap(com.google.common.base.Function<T,ReceivePort<U>> f)

Returns a TransformingReceivePort that receives messages that are transformed by a given flat-mapping function from this channel. Unlike map, the mapping function does not returns a single output message for every input message, but a new ReceivePort. All the returned ports are concatenated into a single ReceivePort that receives the messages received by all the ports in order.

To return a single value the mapping function can make use of Channels.singletonReceivePort(Object). To return a collection, it can make use of Channels.toReceivePort(Iterable). To emit no values, the function can return Channels.emptyReceivePort() or null.

The returned TransformingReceivePort can only be safely used by a single receiver strand.

The returned TransformingReceivePort has the same hashCode as channel and is equal to it.

Parameters:

f - the mapping function

Returns:

a TransformingReceivePort that returns messages that are the result of applying the mapping function to the messages received on the given channel.

take

public TransformingReceivePort<T> take(long count)

Returns a TakeReceivePort that can provide at most count messages from the underlying channel.

The returned TransformingReceivePort has the same hashCode as channel and is equal to it.

Parameters:

count - The maximum number of messages extracted from the underlying channel.

fiberTransform

public <U> TransformingReceivePort<U> fiberTransform(SuspendableAction2<? extends ReceivePort<? super T>,? extends SendPort<? extends U>> transformer,
                                                     Channel<U> out)

Spawns a fiber that transforms values read from this channel and writes values to the out channel.

When the transformation terminates. the output channel is automatically closed. If the transformation terminates abnormally (throws an exception), the output channel is closed with that exception.

Parameters:

out - the output channel

transformer - the transforming operation

Returns:

A TransformingReceivePort wrapping the out channel.

fiberTransform

public <U> TransformingReceivePort<U> fiberTransform(FiberFactory fiberFactory,
                                                     SuspendableAction2<? extends ReceivePort<? super T>,? extends SendPort<? extends U>> transformer,
                                                     Channel<U> out)

Spawns a fiber that transforms values read from this channel and writes values to the out channel.

When the transformation terminates. the output channel is automatically closed. If the transformation terminates abnormally (throws an exception), the output channel is closed with that exception.

Parameters:

fiberFactory - will be used to create the fiber

out - the output channel

transformer - the transforming operation

Returns:

A TransformingReceivePort wrapping the out channel.

forEach

public void forEach(SuspendableAction1<T> action)
             throws SuspendExecution,
                    java.lang.InterruptedException

Performs the given action on each message received by this channel. This method returns only after all messages have been consumed and the channel has been closed.

Parameters:

action -

Throws:

SuspendExecution

java.lang.InterruptedException