class ActorGraphInterpreter extends Actor with ActorLogging

INTERNAL API

final case class Batch [In, Out](max: Long, costFn: (In) ⇒ Long, seed: (In) ⇒ Out, aggregate: (Out, In) ⇒ Out) extends GraphStage[FlowShape[In, Out]] with Product with Serializable

INTERNAL API

final case class Buffer [T](size: Int, overflowStrategy: OverflowStrategy) extends SimpleLinearGraphStage[T] with Product with Serializable

INTERNAL API

final case class Collect [In, Out](pf: PartialFunction[In, Out]) extends GraphStage[FlowShape[In, Out]] with Product with Serializable

INTERNAL API

final class Delay [T] extends SimpleLinearGraphStage[T]

final case class Drop [T](count: Long) extends SimpleLinearGraphStage[T] with Product with Serializable

INTERNAL API

final case class DropWhile [T](p: (T) ⇒ Boolean) extends SimpleLinearGraphStage[T] with Product with Serializable

INTERNAL API

final class DropWithin [T] extends SimpleLinearGraphStage[T]

final class Expand [In, Out] extends GraphStage[FlowShape[In, Out]]

INTERNAL API

final case class Filter [T](p: (T) ⇒ Boolean) extends SimpleLinearGraphStage[T] with Product with Serializable

INTERNAL API

final class FlattenMerge [T, M] extends GraphStage[FlowShape[Graph[SourceShape[T], M], T]]

INTERNAL API

final case class Fold [In, Out](zero: Out, f: (Out, In) ⇒ Out) extends GraphStage[FlowShape[In, Out]] with Product with Serializable

INTERNAL API

final class FoldAsync [In, Out] extends GraphStage[FlowShape[In, Out]]

INTERNAL API

final class GraphInterpreter extends AnyRef

INTERNAL API

From an external viewpoint, the GraphInterpreter takes an assembly of graph processing stages encoded as a GraphInterpreter#GraphAssembly object and provides facilities to execute and interact with this assembly. The lifecycle of the Interpreter is roughly the following:

Boundary logics are attached via attachDownstreamBoundary() and attachUpstreamBoundary()
init() is called
execute() is called whenever there is need for execution, providing an upper limit on the processed events
finish() is called before the interpreter is disposed, preferably after isCompleted returned true, although in abort cases this is not strictly necessary

The execute() method of the interpreter accepts an upper bound on the events it will process. After this limit is reached or there are no more pending events to be processed, the call returns. It is possible to inspect if there are unprocessed events left via the isSuspended method. isCompleted returns true once all stages reported completion inside the interpreter.

The internal architecture of the interpreter is based on the usage of arrays and optimized for reducing allocations on the hot paths.

One of the basic abstractions inside the interpreter is the akka.stream.impl.fusing.GraphInterpreter.Connection. A connection represents an output-input port pair (an analogue for a connected RS Publisher-Subscriber pair). The Connection object contains all the necessary data for the interpreter to pass elements, demand, completion or errors across the Connection. In particular

portStates contains a bitfield that tracks the states of the ports (output-input) corresponding to this connection. This bitfield is used to decode the event that is in-flight.
connectionSlot contains a potential element or exception that accompanies the event encoded in the portStates bitfield
inHandler contains the InHandler instance that handles the events corresponding to the input port of the connection
outHandler contains the OutHandler instance that handles the events corresponding to the output port of the connection

On top of the Connection table there is an eventQueue, represented as a circular buffer of Connections. The queue contains the Connections that have pending events to be processed. The pending event itself is encoded in the portState bitfield of the Connection. This implies that there can be only one event in flight for a given Connection, which is true in almost all cases, except a complete-after-push or fail-after-push which has to be decoded accordingly.

The layout of the portState bitfield is the following:

|- state machn.-| Only one bit is hot among these bits 64 32 16 | 8 4 2 1 | +---+---+---|---+---+---+---| | | | | | | | | | | | | | | From the following flags only one is active in any given time. These bits encode | | | | | | | state machine states, and they are "moved" around using XOR masks to keep other bits | | | | | | | intact. | | | | | | | | | | | | | +- InReady: The input port is ready to be pulled | | | | | +----- Pulling: A pull is active, but have not arrived yet (queued) | | | | +--------- Pushing: A push is active, but have not arrived yet (queued) | | | +------------- OutReady: The output port is ready to be pushed | | | | | +----------------- InClosed: The input port is closed and will not receive any events. | | A push might be still in flight which will be then processed first. | +--------------------- OutClosed: The output port is closed and will not receive any events. +------------------------- InFailed: Always set in conjunction with InClosed. Indicates that the close event is a failure

Sending an event is usually the following sequence:

An action is requested by a stage logic (push, pull, complete, etc.)
the state machine in portStates is transitioned from a ready state to a pending event
the affected Connection is enqueued

Receiving an event is usually the following sequence:

the Connection to be processed is dequeued
the type of the event is determined from the bits set on portStates
the state machine in portStates is transitioned to a ready state
using the inHandlers/outHandlers table the corresponding callback is called on the stage logic.

Because of the FIFO construction of the queue the interpreter is fair, i.e. a pending event is always executed after a bounded number of other events. This property, together with suspendability means that even infinite cycles can be modeled, or even dissolved (if preempted and a "stealing" external event is injected; for example the non-cycle edge of a balance is pulled, dissolving the original cycle).

final class GraphInterpreterShell extends AnyRef

INTERNAL API

final case class GraphModule (assembly: GraphAssembly, shape: Shape, attributes: Attributes, matValIDs: Array[Module]) extends AtomicModule with Product with Serializable

INTERNAL API

final case class GraphStageModule (shape: Shape, attributes: Attributes, stage: GraphStageWithMaterializedValue[Shape, Any]) extends AtomicModule with Product with Serializable

INTERNAL API

final class GroupBy [T, K] extends GraphStage[FlowShape[T, Source[T, NotUsed]]]

INTERNAL API

final case class Grouped [T](n: Int) extends GraphStage[FlowShape[T, Seq[T]]] with Product with Serializable

INTERNAL API

final class GroupedWithin [T] extends GraphStage[FlowShape[T, Seq[T]]]

final case class Intersperse [T](start: Option[T], inject: T, end: Option[T]) extends SimpleLinearGraphStage[T] with Product with Serializable

INTERNAL API

final case class LimitWeighted [T](n: Long, costFn: (T) ⇒ Long) extends SimpleLinearGraphStage[T] with Product with Serializable

INTERNAL API

final case class Log [T](name: String, extract: (T) ⇒ Any, logAdapter: Option[LoggingAdapter]) extends SimpleLinearGraphStage[T] with Product with Serializable

INTERNAL API

final case class Map [In, Out](f: (In) ⇒ Out) extends GraphStage[FlowShape[In, Out]] with Product with Serializable

INTERNAL API

final case class MapAsync [In, Out](parallelism: Int, f: (In) ⇒ Future[Out]) extends GraphStage[FlowShape[In, Out]] with Product with Serializable

INTERNAL API

final case class MapAsyncUnordered [In, Out](parallelism: Int, f: (In) ⇒ Future[Out]) extends GraphStage[FlowShape[In, Out]] with Product with Serializable

INTERNAL API

final case class MapError [T](f: PartialFunction[Throwable, Throwable]) extends SimpleLinearGraphStage[T] with Product with Serializable

Maps error with the provided function if it is defined for an error or, otherwise, passes it on unchanged.

While similar to Recover this stage can be used to transform an error signal to a different one *without* logging it as an error in the process. So in that sense it is NOT exactly equivalent to recover(t => throw t2) since recover would log the t2 error.

final class PrefixAndTail [T] extends GraphStage[FlowShape[T, (Seq[T], Source[T, NotUsed])]]

INTERNAL API

final case class Recover [T](pf: PartialFunction[Throwable, T]) extends SimpleLinearGraphStage[T] with Product with Serializable

INTERNAL API

final class RecoverWith [T, M] extends SimpleLinearGraphStage[T]

final class Reduce [T] extends SimpleLinearGraphStage[T]

INTERNAL API

final case class Scan [In, Out](zero: Out, f: (Out, In) ⇒ Out) extends GraphStage[FlowShape[In, Out]] with Product with Serializable

INTERNAL API

final case class ScanAsync [In, Out](zero: Out, f: (Out, In) ⇒ Future[Out]) extends GraphStage[FlowShape[In, Out]] with Product with Serializable

INTERNAL API

final case class Sliding [T](n: Int, step: Int) extends GraphStage[FlowShape[T, Seq[T]]] with Product with Serializable

INTERNAL API

final class Split [T] extends GraphStage[FlowShape[T, Source[T, NotUsed]]]

INTERNAL API

final class StatefulMapConcat [In, Out] extends GraphStage[FlowShape[In, Out]]

INTERNAL API

final class SubSource [T] extends GraphStage[SourceShape[T]]

INTERNAL API

abstract class SupervisedGraphStageLogic extends GraphStageLogic

INTERNAL API

final case class Take [T](count: Long) extends SimpleLinearGraphStage[T] with Product with Serializable

INTERNAL API

final case class TakeWhile [T](p: (T) ⇒ Boolean, inclusive: Boolean = false) extends SimpleLinearGraphStage[T] with Product with Serializable

INTERNAL API

final class TakeWithin [T] extends SimpleLinearGraphStage[T]

Packages

fusing

package fusing

Type Members

Value Members

Ungrouped

Packages

fusing 

package fusing

Type Members

Value Members

Ungrouped

fusing