0
# Akka Streams
1
2
Akka Streams is a reactive streams implementation for processing and transferring sequences of elements using bounded buffer space. This library provides a domain-specific language for expressing complex data transformation pipelines with automatic backpressure management, enabling developers to build robust, asynchronous stream processing applications.
3
4
## Package Information
5
6
- **Package Name**: com.typesafe.akka:akka-stream_2.13.0-M5
7
- **Package Type**: maven
8
- **Language**: Scala (with Java API)
9
- **Version**: 2.5.23
10
- **Installation**: Add to `build.sbt`: `libraryDependencies += "com.typesafe.akka" %% "akka-stream" % "2.5.23"`
11
12
## Core Imports
13
14
```scala
15
import akka.stream.scaladsl.{Source, Flow, Sink}
16
import akka.stream.{ActorMaterializer, Materializer}
17
import akka.actor.ActorSystem
18
import akka.{Done, NotUsed}
19
```
20
21
Java API:
22
```java
23
import akka.stream.javadsl.*;
24
import akka.stream.ActorMaterializer;
25
import akka.stream.Materializer;
26
import akka.actor.ActorSystem;
27
```
28
29
## Basic Usage
30
31
```scala
32
import akka.actor.ActorSystem
33
import akka.stream.ActorMaterializer
34
import akka.stream.scaladsl.{Source, Sink}
35
36
implicit val system = ActorSystem("MySystem")
37
implicit val materializer: ActorMaterializer = ActorMaterializer()
38
39
// Create a simple stream: numbers 1 to 10, multiply by 2, print results
40
val source: Source[Int, NotUsed] = Source(1 to 10)
41
val sink: Sink[Any, Future[Done]] = Sink.foreach(println)
42
43
val result: Future[Done] = source
44
.map(_ * 2)
45
.runWith(sink)
46
47
// Result will print: 2, 4, 6, 8, 10, 12, 14, 16, 18, 20
48
```
49
50
## Architecture
51
52
Akka Streams is built around several key abstractions:
53
54
- **Graph**: The blueprint of a stream processing topology
55
- **Shape**: Defines the inlets and outlets of a graph component
56
- **Materializer**: Responsible for turning graph blueprints into running streams
57
- **Source**: Stream component with one output (data producer)
58
- **Flow**: Stream component with one input and one output (data transformer)
59
- **Sink**: Stream component with one input (data consumer)
60
61
All stream processing is built using these composable, type-safe building blocks that automatically handle backpressure according to the Reactive Streams specification.
62
63
## Capabilities
64
65
### Core Stream Components
66
67
The fundamental building blocks for creating reactive streams with type-safe composition and automatic backpressure handling.
68
69
```scala { .api }
70
// Source - produces elements
71
final class Source[+Out, +Mat](traversalBuilder: LinearTraversalBuilder, shape: SourceShape[Out])
72
73
// Flow - transforms elements
74
final class Flow[-In, +Out, +Mat](traversalBuilder: LinearTraversalBuilder, shape: FlowShape[In, Out])
75
76
// Sink - consumes elements
77
final class Sink[-In, +Mat](traversalBuilder: LinearTraversalBuilder, shape: SinkShape[In])
78
```
79
80
[Core Stream Components](./core-components.md)
81
82
### Graph Building and Composition
83
84
Advanced graph construction using GraphDSL for complex stream topologies including fan-in, fan-out, and custom shapes.
85
86
```scala { .api }
87
object GraphDSL {
88
def create[S <: Shape, Mat](buildBlock: GraphDSL.Builder[Mat] => S): Graph[S, Mat]
89
90
class Builder[+M] {
91
def add[S <: Shape](graph: Graph[S, _]): S
92
// Connection operators: ~>, <~, via, to, from
93
}
94
}
95
```
96
97
[Graph Building and Composition](./graph-building.md)
98
99
### Stream Operations and Transformations
100
101
Comprehensive set of stream processing operations including mapping, filtering, grouping, timing, and error handling.
102
103
```scala { .api }
104
trait FlowOps[+Out, +Mat] {
105
def map[T](f: Out => T): Repr[T]
106
def filter(p: Out => Boolean): Repr[Out]
107
def mapAsync[T](parallelism: Int)(f: Out => Future[T]): Repr[T]
108
def groupBy[K](maxSubstreams: Int, f: Out => K): SubFlow[Out, Mat, Repr, Closed]
109
def throttle(elements: Int, per: FiniteDuration): Repr[Out]
110
}
111
```
112
113
[Stream Operations and Transformations](./stream-operations.md)
114
115
### Materialization and Execution
116
117
Stream materialization with ActorMaterializer, lifecycle management, and execution control.
118
119
```scala { .api }
120
abstract class Materializer {
121
def materialize[Mat](runnable: Graph[ClosedShape, Mat]): Mat
122
def withNamePrefix(name: String): Materializer
123
}
124
125
object ActorMaterializer {
126
def apply()(implicit context: ActorRefFactory): ActorMaterializer
127
}
128
```
129
130
[Materialization and Execution](./materialization.md)
131
132
### Junction Operations
133
134
Stream junction operators for merging, broadcasting, zipping, and partitioning multiple streams.
135
136
```scala { .api }
137
class Merge[T](inputPorts: Int, eagerComplete: Boolean)
138
class Broadcast[T](outputPorts: Int, eagerCancel: Boolean)
139
class Zip[A, B] extends ZipWith2[A, B, (A, B)]
140
class Partition[T](outputPorts: Int, partitioner: T => Int)
141
```
142
143
[Junction Operations](./junction-operations.md)
144
145
### I/O Integration
146
147
File I/O, TCP networking, and integration with Java streams and other I/O systems.
148
149
```scala { .api }
150
object FileIO {
151
def fromPath(path: Path, chunkSize: Int = 8192): Source[ByteString, Future[IOResult]]
152
def toPath(path: Path): Sink[ByteString, Future[IOResult]]
153
}
154
155
object Tcp {
156
def outgoingConnection(remoteAddress: InetSocketAddress): Flow[ByteString, ByteString, Future[OutgoingConnection]]
157
}
158
```
159
160
[I/O Integration](./io-integration.md)
161
162
### Error Handling and Supervision
163
164
Comprehensive error handling with supervision strategies, recovery operations, and stream resilience patterns.
165
166
```scala { .api }
167
object Supervision {
168
sealed trait Directive
169
case object Stop extends Directive
170
case object Resume extends Directive
171
case object Restart extends Directive
172
173
type Decider = Function[Throwable, Directive]
174
}
175
```
176
177
[Error Handling and Supervision](./error-handling.md)
178
179
### Control Flow and Lifecycle
180
181
Stream lifecycle management with KillSwitch, StreamRefs for distribution, and queue integration.
182
183
```scala { .api }
184
trait KillSwitch {
185
def shutdown(): Unit
186
def abort(ex: Throwable): Unit
187
}
188
189
object KillSwitches {
190
def single[T]: Graph[FlowShape[T, T], UniqueKillSwitch]
191
def shared(name: String): SharedKillSwitch
192
}
193
```
194
195
[Control Flow and Lifecycle](./control-flow.md)
196
197
## Types
198
199
```scala { .api }
200
// Fundamental types
201
type NotUsed = akka.NotUsed.type
202
type Done = akka.Done.type
203
204
// Shape hierarchy
205
abstract class Shape {
206
def inlets: immutable.Seq[Inlet[_]]
207
def outlets: immutable.Seq[Outlet[_]]
208
}
209
210
case class SourceShape[+T](out: Outlet[T]) extends Shape
211
case class FlowShape[-I, +O](in: Inlet[I], out: Outlet[O]) extends Shape
212
case class SinkShape[-T](in: Inlet[T]) extends Shape
213
214
// Ports
215
final class Inlet[T](s: String)
216
final class Outlet[T](s: String)
217
218
// Materialization
219
trait Graph[+S <: Shape, +M] {
220
def shape: S
221
def withAttributes(attr: Attributes): Graph[S, M]
222
}
223
224
// Results
225
case class IOResult(count: Long, status: Try[Done]) {
226
def wasSuccessful: Boolean
227
}
228
229
// Queue integration
230
sealed abstract class QueueOfferResult
231
object QueueOfferResult {
232
case object Enqueued extends QueueOfferResult
233
case object Dropped extends QueueOfferResult
234
case class Failure(cause: Throwable) extends QueueOfferResult
235
case object QueueClosed extends QueueOfferResult
236
}
237
238
// Strategies
239
sealed abstract class OverflowStrategy
240
object OverflowStrategy {
241
def dropHead: OverflowStrategy
242
def dropTail: OverflowStrategy
243
def backpressure: OverflowStrategy
244
def fail: OverflowStrategy
245
}
246
247
// Attributes
248
final case class Attributes(attributeList: List[Attributes.Attribute]) {
249
def and(other: Attributes): Attributes
250
def get[T <: Attributes.Attribute: ClassTag]: Option[T]
251
}
252
```