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 +95,23 @@ To update the write side, we added `.interleave(Stream.constant("\n"))` before d
95
95
96
96
#### Handling Connection Errors
97
97
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.
98
+
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.
129
+
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
130
131
131
### Servers
132
132
@@ -136,18 +136,20 @@ Now let's implement a server application that communicates with the client app w
.handleErrorWith(_ =>Stream.empty) // handle errors of client sockets
148
+
}
147
149
}.parJoin(100).compile.drain
148
150
```
149
151
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.
152
+
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
153
152
154
We map over this infinite stream of clients and provide the logic for handling an individual client. In this case,
153
155
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 +158,7 @@ Since we mapped over the infinite client stream, we end up with a `Stream[F, Str
156
158
157
159
In joining all these streams together, be prudent to handle errors in the client streams.
158
160
159
-
The pattern of `Network[F].server(address).map(handleClient).parJoin(maxConcurrentClients)` is very common when working with server sockets.
161
+
The pattern of `Network[F].bind(address).map(ss => handleClient(ss.accept)).parJoin(maxConcurrentClients)` is very common when working with server sockets.
160
162
161
163
A simpler echo server could be implemented with this core logic:
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