You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
To open a socket that's connected to `localhost:5555`, we use the `client` method on the `Network` capability. The `Network` capability provides the runtime environment for the sockets it creates.
44
+
To open a socket that's connected to `localhost:5555`, we use the `connect` method on the `Network` capability. The `Network` capability provides the runtime environment for the sockets it creates.
45
45
46
-
The `Network[F].client` method returns a `Resource[F, Socket[F]]` which automatically closes the socket after the resource has been used. To write data to the socket, we call `socket.write`, which takes a `Chunk[Byte]` and returns an `F[Unit]`. Once the write completes, we do a single read from the socket via `socket.read`, passing the maximum amount of bytes we want to read. The returns an `F[Option[Chunk[Byte]]]` -- `None` if the socket reaches end of input and `Some` if the read produced a chunk. Finally, we print the response to the console.
46
+
The `Network[F].connect` method returns a `Resource[F, Socket[F]]` which automatically closes the socket after the resource has been used. To write data to the socket, we call `socket.write`, which takes a `Chunk[Byte]` and returns an `F[Unit]`. Once the write completes, we do a single read from the socket via `socket.read`, passing the maximum amount of bytes we want to read. The returns an `F[Option[Chunk[Byte]]]` -- `None` if the socket reaches end of input and `Some` if the read produced a chunk. Finally, we print the response to the console.
47
47
48
48
Note we aren't doing any binary message framing or packetization in this example. Hence, it's very possible for the single read to only receive a portion of the original message -- perhaps just the bytes for `"Hello, w"`. We can use FS2 streams to simplify this. The `Socket` trait defines stream operations -- `writes` and `reads`. We could rewrite this example using the stream operations like so:
@@ -95,23 +94,23 @@ To update the write side, we added `.interleave(Stream.constant("\n"))` before d
95
94
96
95
#### Handling Connection Errors
97
96
98
-
If a TCP connection cannot be established, `socketGroup.client` fails with a `java.net.ConnectException`. To automatically attempt a reconnection, we can handle the `ConnectException` and try connecting again.
97
+
If a TCP connection cannot be established, `connect` fails with a `fs2.io.net.ConnectException`. To automatically attempt a reconnection, we can handle the `ConnectException` and try connecting again.
We've extracted the `Network[IO].client` call in to a new method called `connect`. The connect method attempts to create a client and handles the `ConnectException`. Upon encountering the exception, we call `connect` recursively after a 5 second delay. Because we are using `delayBy`, we needed to add a `Temporal` constraint to `F`. This same pattern could be used for more advanced retry strategies -- e.g., exponential delays and failing after a fixed number of attempts. Streams that call methods on `Socket` can fail with exceptions due to loss of the underlying TCP connection. Such exceptions can be handled in a similar manner.
128
+
We've extracted the `Network[IO].connect` call in to a new method called `retryingConnect`. The `retryingConnect` method attempts to create a client and handles the `ConnectException`. Upon encountering the exception, we call `retryingConnect` recursively after a 5 second delay. Because we are using `delayBy`, we needed to add a `Temporal` constraint to `F`. This same pattern could be used for more advanced retry strategies -- e.g., exponential delays and failing after a fixed number of attempts. Streams that call methods on `Socket` can fail with exceptions due to loss of the underlying TCP connection. Such exceptions can be handled in a similar manner.
130
129
131
130
### Servers
132
131
@@ -136,18 +135,20 @@ Now let's implement a server application that communicates with the client app w
.handleErrorWith(_ =>Stream.empty) // handle errors of client sockets
147
+
}
147
148
}.parJoin(100).compile.drain
148
149
```
149
150
150
-
We start with a call to `Network[IO].server` which returns a value of an interesting type --`Stream[F, Socket[F]]`. This is an infinite stream of client sockets -- each time a client connects to the server, a `Socket[F]` is emitted, allowing interaction with that client. The lifetime of the client socket is managed by the overall stream -- e.g. flat mapping over a socket will keep that socket open until the returned inner stream completes, at which point, the client socket is closed and any underlying resources are returned to the runtime environment.
151
+
We start with a call to `Network[IO].bind` which returns a value of type `Resource[F, ServerSocket[F]]`. The `ServerSocket`type defines an `accept` method of type`Stream[F, Socket[F]]`. This is an infinite stream of client sockets -- each time a client connects to the server, a `Socket[F]` is emitted, allowing interaction with that client. The lifetime of the client socket is managed by the overall stream -- e.g. flat mapping over a socket will keep that socket open until the returned inner stream completes, at which point, the client socket is closed and any underlying resources are returned to the runtime environment.
151
152
152
153
We map over this infinite stream of clients and provide the logic for handling an individual client. In this case,
153
154
we read from the client socket, UTF-8 decode the received bytes, extract individual lines, and then write each line back to the client. This logic is implemented as a single `Stream[F, Unit]`.
@@ -156,7 +157,7 @@ Since we mapped over the infinite client stream, we end up with a `Stream[F, Str
156
157
157
158
In joining all these streams together, be prudent to handle errors in the client streams.
158
159
159
-
The pattern of `Network[F].server(address).map(handleClient).parJoin(maxConcurrentClients)` is very common when working with server sockets.
160
+
The pattern of `Network[F].bind(address).map(ss => handleClient(ss.accept)).parJoin(maxConcurrentClients)` is very common when working with server sockets.
160
161
161
162
A simpler echo server could be implemented with this core logic:
162
163
@@ -170,26 +171,24 @@ The [fs2-chat](https://github.com/functional-streams-for-scala/fs2-chat) sample
170
171
171
172
## UDP
172
173
173
-
UDP support works much the same way as TCP. The `fs2.io.net.DatagramSocket` trait provides mechanisms for reading and writing UDP datagrams. UDP sockets are created via the `openDatagramSocket` method on `fs2.io.net.Network`. Unlike TCP, there's no differentiation between client and server sockets. Additionally, since UDP is a packet based protocol, read and write operations use `fs2.io.net.Datagram` values, which consist of a `Chunk[Byte]` and a `SocketAddress[IpAddress]`.
174
+
UDP support works much the same way as TCP. The `fs2.io.net.DatagramSocket` trait provides mechanisms for reading and writing UDP datagrams. UDP sockets are created via the `bindDatagramSocket` method on `fs2.io.net.Network`. Unlike TCP, there's no differentiation between client and server sockets.
174
175
175
176
Adapting the TCP client example for UDP gives us the following:
When writing, we map each chunk of bytes to a `Datagram`, which includes the destination address of the packet. When reading, we convert the `Stream[F, Datagram]` to a `Stream[F, Byte]` via `flatMap(datagram => Stream.chunk(datagram.bytes))`. Otherwise, the example is unchanged.
202
+
We call `socket.write`, supplying a chunk of bytes and the destination address. When reading, we convert the `Stream[F, Datagram]` to a `Stream[F, Byte]` via `flatMap(datagram => Stream.chunk(datagram.bytes))`. Otherwise, the example is unchanged.
Each emitted sample is the sum of bits received during each one second period. Let's compute an average of that value over the last 10 seconds. We can do this via `mapAccumulate` along with a `scala.collection.immutable.Queue`:
0 commit comments