diff --git a/io/js/src/main/scala/fs2/io/ioplatform.scala b/io/js/src/main/scala/fs2/io/ioplatform.scala index e6b635875b..d071788759 100644 --- a/io/js/src/main/scala/fs2/io/ioplatform.scala +++ b/io/js/src/main/scala/fs2/io/ioplatform.scala @@ -192,17 +192,25 @@ private[fs2] trait ioplatform { def writeWritable[F[_]]( writable: F[Writable], endAfterUse: Boolean = true - )(implicit F: Async[F]): Pipe[F, Byte, Nothing] = + )(implicit F: Async[F]): Pipe[F, Byte, Nothing] = stream => + writeWritableIncremental(writable, endAfterUse)(F)(stream).drain + + /** Writes all bytes to the specified `Writable`. + */ + def writeWritableIncremental[F[_]]( + writable: F[Writable], + endAfterUse: Boolean = true + )(implicit F: Async[F]): Pipe[F, Byte, Int] = in => Stream .eval(writable) .flatMap { writable => - val writes = in.chunks.foreach { chunk => - F.async[Unit] { cb => + val writes = in.chunks.evalMap { chunk => + F.async[Int] { cb => F.delay { writable.write( chunk.toUint8Array, - e => cb(e.filterNot(_ == null).toLeft(()).leftMap(js.JavaScriptException)) + e => cb(e.filterNot(_ == null).toLeft(chunk.size).leftMap(js.JavaScriptException)) ) Some(F.delay(writable.destroy())) } diff --git a/io/js/src/main/scala/fs2/io/net/SocketPlatform.scala b/io/js/src/main/scala/fs2/io/net/SocketPlatform.scala index 52d7424d18..36abee72f8 100644 --- a/io/js/src/main/scala/fs2/io/net/SocketPlatform.scala +++ b/io/js/src/main/scala/fs2/io/net/SocketPlatform.scala @@ -112,8 +112,15 @@ private[net] trait SocketCompanionPlatform { override def write(bytes: Chunk[Byte]): F[Unit] = Stream.chunk(bytes).through(writes).compile.drain + override def writeIncremental(bytes: fs2.Chunk[Byte]): fs2.Stream[F, Int] = + Stream.chunk(bytes).through(writesIncremental) + override def writes: Pipe[F, Byte, Nothing] = writeWritable(F.pure(sock), endAfterUse = false) + + override def writesIncremental: Pipe[F, Byte, Int] = + writeWritableIncremental(F.pure(sock), endAfterUse = false) + } } diff --git a/io/jvm-native/src/main/scala/fs2/io/net/SocketPlatform.scala b/io/jvm-native/src/main/scala/fs2/io/net/SocketPlatform.scala index f42cfd7a2d..4ca73a6ee5 100644 --- a/io/jvm-native/src/main/scala/fs2/io/net/SocketPlatform.scala +++ b/io/jvm-native/src/main/scala/fs2/io/net/SocketPlatform.scala @@ -148,6 +148,22 @@ private[net] trait SocketCompanionPlatform { } } + override def writeIncremental(bytes: Chunk[Byte]): Stream[F, Int] = { + def go(buff: ByteBuffer): Pull[F, Int, Unit] = + Pull + .eval(F.async[Int] { cb => + ch.write(buff, null, new IntCompletionHandler(cb)) + F.delay(Some(endOfOutput.voidError)) + }) + .flatMap { written => + Pull.output1(written) >> + go(buff).whenA(written >= 0 && buff.remaining() > 0) + } + + Stream.resource(writeMutex.lock) >> + Stream.eval(F.delay(bytes.toByteBuffer)).flatMap(go(_).stream) + } + override def localAddress: F[SocketAddress[IpAddress]] = asyncInstance.pure(address.asIpUnsafe) diff --git a/io/jvm/src/main/scala/fs2/io/net/AsyncUnixSocketsProvider.scala b/io/jvm/src/main/scala/fs2/io/net/AsyncUnixSocketsProvider.scala index 11aa99ead6..ea23530c88 100644 --- a/io/jvm/src/main/scala/fs2/io/net/AsyncUnixSocketsProvider.scala +++ b/io/jvm/src/main/scala/fs2/io/net/AsyncUnixSocketsProvider.scala @@ -117,6 +117,16 @@ private[net] object AsyncUnixSocketsProvider { } } + override def writeIncremental(bytes: Chunk[Byte]): Stream[F, Int] = { + def go(buff: ByteBuffer): Pull[F, Int, Unit] = + Pull + .eval(evalOnVirtualThreadIfAvailable(F.blocking(ch.write(buff))).cancelable(close)) + .flatMap(written => Pull.output1(written) >> go(buff).whenA(buff.remaining() > 0)) + + Stream.resource(writeMutex.lock) *> + Stream.eval(F.delay(bytes.toByteBuffer)).flatMap(go(_).stream) + } + private def raiseIpAddressError[A]: F[A] = F.raiseError(new UnsupportedOperationException("Unix sockets do not use IP addressing")) diff --git a/io/jvm/src/main/scala/fs2/io/net/SelectingSocket.scala b/io/jvm/src/main/scala/fs2/io/net/SelectingSocket.scala index 2b7cf81031..f5f23d7c87 100644 --- a/io/jvm/src/main/scala/fs2/io/net/SelectingSocket.scala +++ b/io/jvm/src/main/scala/fs2/io/net/SelectingSocket.scala @@ -77,6 +77,20 @@ private final class SelectingSocket[F[_]: LiftIO] private ( } } + override def writeIncremental(bytes: Chunk[Byte]): Stream[F, Int] = { + def go(buf: ByteBuffer): Pull[F, Int, Unit] = + Pull.eval(F.delay(ch.write(buf))).flatMap { written => + if (buf.remaining() > 0) + Pull.eval(selector.select(ch, OP_WRITE).to).void >> + Pull.output1(written) >> + go(buf) + else Pull.output1(written) + } + + Stream.resource(writeMutex.lock) >> + Stream.eval(F.delay(bytes.toByteBuffer)).flatMap(go(_).stream) + } + def isOpen: F[Boolean] = F.delay(ch.isOpen) def endOfOutput: F[Unit] = diff --git a/io/jvm/src/main/scala/fs2/io/net/tls/TLSContextPlatform.scala b/io/jvm/src/main/scala/fs2/io/net/tls/TLSContextPlatform.scala index 8d301331e9..4bfc914667 100644 --- a/io/jvm/src/main/scala/fs2/io/net/tls/TLSContextPlatform.scala +++ b/io/jvm/src/main/scala/fs2/io/net/tls/TLSContextPlatform.scala @@ -34,7 +34,6 @@ import javax.net.ssl.{ TrustManagerFactory, X509ExtendedTrustManager } -import cats.Applicative import cats.effect.kernel.{Async, Resource} import cats.syntax.all._ import com.comcast.ip4s.{IpAddress, SocketAddress} @@ -159,8 +158,8 @@ private[tls] trait TLSContextCompanionPlatform { self: TLSContext.type => .eval( engine( new TLSEngine.Binding[F] { - def write(data: Chunk[Byte]): F[Unit] = - socket.write(data) + def write(data: Chunk[Byte]): Stream[F, Int] = + socket.writeIncremental(data) def read(maxBytes: Int): F[Option[Chunk[Byte]]] = socket.read(maxBytes) }, @@ -194,9 +193,9 @@ private[tls] trait TLSContextCompanionPlatform { self: TLSContext.type => .eval( engine( new TLSEngine.Binding[F] { - def write(data: Chunk[Byte]): F[Unit] = - if (data.isEmpty) Applicative[F].unit - else socket.write(data, remoteAddress) + def write(data: Chunk[Byte]): Stream[F, Int] = + if (data.isEmpty) Stream.empty + else Stream.eval(socket.write(data, remoteAddress).as(data.size)) def read(maxBytes: Int): F[Option[Chunk[Byte]]] = socket.read.map(p => Some(p.bytes)) }, diff --git a/io/jvm/src/main/scala/fs2/io/net/tls/TLSEngine.scala b/io/jvm/src/main/scala/fs2/io/net/tls/TLSEngine.scala index cb7a4bdf23..36823d2185 100644 --- a/io/jvm/src/main/scala/fs2/io/net/tls/TLSEngine.scala +++ b/io/jvm/src/main/scala/fs2/io/net/tls/TLSEngine.scala @@ -42,12 +42,13 @@ private[tls] trait TLSEngine[F[_]] { def stopWrap: F[Unit] def stopUnwrap: F[Unit] def write(data: Chunk[Byte]): F[Unit] + def writeIncremental(data: Chunk[Byte]): Stream[F, Int] def read(maxBytes: Int): F[Option[Chunk[Byte]]] } private[tls] object TLSEngine { trait Binding[F[_]] { - def write(data: Chunk[Byte]): F[Unit] + def write(data: Chunk[Byte]): Stream[F, Int] def read(maxBytes: Int): F[Option[Chunk[Byte]]] } @@ -85,41 +86,54 @@ private[tls] object TLSEngine { def stopUnwrap = Sync[F].delay(engine.closeInbound()).attempt.void def write(data: Chunk[Byte]): F[Unit] = - writeMutex.lock.surround(write0(data)) + writeIncremental(data).compile.drain - private def write0(data: Chunk[Byte]): F[Unit] = - wrapBuffer.input(data) >> wrap + def writeIncremental(data: Chunk[Byte]): Stream[F, Int] = + Stream.resource(writeMutex.lock) >> write0(data) + + private def write0(data: Chunk[Byte]): Stream[F, Int] = + Stream.eval(wrapBuffer.input(data)) >> wrap.stream /** Performs a wrap operation on the underlying engine. */ - private def wrap: F[Unit] = - wrapBuffer - .perform(engine.wrap(_, _)) - .flatTap(result => log(s"wrap result: $result")) + private def wrap: Pull[F, Int, Unit] = + Pull + .eval( + wrapBuffer + .perform(engine.wrap(_, _)) + .flatTap(result => log(s"wrap result: $result")) + ) .flatMap { result => result.getStatus match { case SSLEngineResult.Status.OK => - doWrite >> { + doWrite.pull.echo >> { result.getHandshakeStatus match { case SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING => - wrapBuffer.inputRemains + Pull + .eval(wrapBuffer.inputRemains) .flatMap(x => wrap.whenA(x > 0 && result.bytesConsumed > 0)) case _ => - handshakeMutex.lock - .surround(stepHandshake(result, true)) >> wrap + // TODO: is this the right way to use and close the resource? + Pull.scope( + Stream + .resource(handshakeMutex.lock) + .flatMap(_ => stepHandshake(result, true).stream) + .pull + .echo + ) >> wrap } } case SSLEngineResult.Status.BUFFER_UNDERFLOW => - doWrite + doWrite.pull.echo case SSLEngineResult.Status.BUFFER_OVERFLOW => - wrapBuffer.expandOutput >> wrap + Pull.eval(wrapBuffer.expandOutput) >> wrap case SSLEngineResult.Status.CLOSED => - stopWrap + Pull.eval(stopWrap) } } - private def doWrite: F[Unit] = - wrapBuffer.output(Int.MaxValue).flatMap { out => - if (out.isEmpty) Applicative[F].unit + private def doWrite: Stream[F, Int] = + Stream.eval(wrapBuffer.output(Int.MaxValue)).flatMap { out => + if (out.isEmpty) Stream.empty else binding.write(out) } @@ -169,7 +183,7 @@ private[tls] object TLSEngine { unwrap(maxBytes) case _ => handshakeMutex.lock - .surround(stepHandshake(result, false)) >> unwrap( + .surround(stepHandshake(result, false).stream.compile.drain) >> unwrap( maxBytes ) } @@ -194,28 +208,29 @@ private[tls] object TLSEngine { private def stepHandshake( result: SSLEngineResult, lastOperationWrap: Boolean - ): F[Unit] = + ): Pull[F, Int, Unit] = result.getHandshakeStatus match { case SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING => - Applicative[F].unit + Pull.done case SSLEngineResult.HandshakeStatus.FINISHED => - unwrapBuffer.inputRemains.flatMap { remaining => + Pull.eval(unwrapBuffer.inputRemains).flatMap { remaining => if (remaining > 0) unwrapHandshake - else Applicative[F].unit + else Pull.done } case SSLEngineResult.HandshakeStatus.NEED_TASK => - sslEngineTaskRunner.runDelegatedTasks >> (if (lastOperationWrap) wrapHandshake - else unwrapHandshake) + Pull.eval(sslEngineTaskRunner.runDelegatedTasks) >> + (if (lastOperationWrap) wrapHandshake + else unwrapHandshake) case SSLEngineResult.HandshakeStatus.NEED_WRAP => wrapHandshake case SSLEngineResult.HandshakeStatus.NEED_UNWRAP => - unwrapBuffer.inputRemains.flatMap { remaining => + Pull.eval(unwrapBuffer.inputRemains).flatMap { remaining => if (remaining > 0 && result.getStatus != SSLEngineResult.Status.BUFFER_UNDERFLOW) unwrapHandshake else - binding.read(engine.getSession.getPacketBufferSize).flatMap { - case Some(c) => unwrapBuffer.input(c) >> unwrapHandshake - case None => stopUnwrap + Pull.eval(binding.read(engine.getSession.getPacketBufferSize)).flatMap { + case Some(c) => Pull.eval(unwrapBuffer.input(c)) >> unwrapHandshake + case None => Pull.eval(stopUnwrap) } } case SSLEngineResult.HandshakeStatus.NEED_UNWRAP_AGAIN => @@ -223,29 +238,35 @@ private[tls] object TLSEngine { } /** Performs a wrap operation as part of handshaking. */ - private def wrapHandshake: F[Unit] = - wrapBuffer - .perform(engine.wrap(_, _)) - .flatTap(result => log(s"wrapHandshake result: $result")) + private def wrapHandshake: Pull[F, Int, Unit] = + Pull + .eval( + wrapBuffer + .perform(engine.wrap(_, _)) + .flatTap(result => log(s"wrapHandshake result: $result")) + ) .flatMap { result => result.getStatus match { case SSLEngineResult.Status.OK | SSLEngineResult.Status.BUFFER_UNDERFLOW => - doWrite >> stepHandshake( + doWrite.pull.echo >> stepHandshake( result, true ) case SSLEngineResult.Status.BUFFER_OVERFLOW => - wrapBuffer.expandOutput >> wrapHandshake + Pull.eval(wrapBuffer.expandOutput) >> wrapHandshake case SSLEngineResult.Status.CLOSED => - stopWrap >> stopUnwrap + Pull.eval(stopWrap >> stopUnwrap) } } /** Performs an unwrap operation as part of handshaking. */ - private def unwrapHandshake: F[Unit] = - unwrapBuffer - .perform(engine.unwrap(_, _)) - .flatTap(result => log(s"unwrapHandshake result: $result")) + private def unwrapHandshake: Pull[F, Int, Unit] = + Pull + .eval( + unwrapBuffer + .perform(engine.unwrap(_, _)) + .flatTap(result => log(s"unwrapHandshake result: $result")) + ) .flatMap { result => result.getStatus match { case SSLEngineResult.Status.OK => @@ -253,9 +274,9 @@ private[tls] object TLSEngine { case SSLEngineResult.Status.BUFFER_UNDERFLOW => stepHandshake(result, false) case SSLEngineResult.Status.BUFFER_OVERFLOW => - unwrapBuffer.expandOutput >> unwrapHandshake + Pull.eval(unwrapBuffer.expandOutput) >> unwrapHandshake case SSLEngineResult.Status.CLOSED => - stopWrap >> stopUnwrap + Pull.eval(stopWrap >> stopUnwrap) } } } diff --git a/io/jvm/src/main/scala/fs2/io/net/tls/TLSSocketPlatform.scala b/io/jvm/src/main/scala/fs2/io/net/tls/TLSSocketPlatform.scala index 5bca05195f..6c7c627d2c 100644 --- a/io/jvm/src/main/scala/fs2/io/net/tls/TLSSocketPlatform.scala +++ b/io/jvm/src/main/scala/fs2/io/net/tls/TLSSocketPlatform.scala @@ -56,6 +56,9 @@ private[tls] trait TLSSocketCompanionPlatform { self: TLSSocket.type => def write(bytes: Chunk[Byte]): F[Unit] = engine.write(bytes) + override def writeIncremental(bytes: Chunk[Byte]): Stream[F, Int] = + engine.writeIncremental(bytes) + private def read0(maxBytes: Int): F[Option[Chunk[Byte]]] = engine.read(maxBytes) diff --git a/io/jvm/src/test/scala/fs2/io/net/tls/TLSSocketSuite.scala b/io/jvm/src/test/scala/fs2/io/net/tls/TLSSocketSuite.scala index 459f2d0adf..1690145ea6 100644 --- a/io/jvm/src/test/scala/fs2/io/net/tls/TLSSocketSuite.scala +++ b/io/jvm/src/test/scala/fs2/io/net/tls/TLSSocketSuite.scala @@ -251,6 +251,18 @@ class TLSSocketSuite extends TLSSuite { raw.write(b) >> IO(totalWritten += b.size) >> IO(totalWritten >= limit) .ifM(endOfOutput, IO.unit) } + + override def writeIncremental(bytes: Chunk[Byte]): Stream[IO, Int] = + if (totalWritten >= limit) Stream.eval(endOfOutput).drain + else { + val b = bytes.take(limit - totalWritten) + raw.writeIncremental(b) ++ Stream + .eval( + IO(totalWritten += b.size) >> IO(totalWritten >= limit).ifM(endOfOutput, IO.unit) + ) + .drain + } + } // Setup an HTTPS echo server & a client that starts a TLS handshake but only sends the first few bytes and then signals no more output diff --git a/io/native/src/main/scala/fs2/io/net/FdPollingSocket.scala b/io/native/src/main/scala/fs2/io/net/FdPollingSocket.scala index 939a936a10..a34d2786cb 100644 --- a/io/native/src/main/scala/fs2/io/net/FdPollingSocket.scala +++ b/io/native/src/main/scala/fs2/io/net/FdPollingSocket.scala @@ -23,7 +23,6 @@ package fs2 package io.net import cats.effect.{Async, FileDescriptorPollHandle, IO, LiftIO, Resource} -import cats.syntax.all._ import com.comcast.ip4s.GenSocketAddress import fs2.io.internal.NativeUtil._ import fs2.io.internal.{ResizableBuffer, SocketHelpers} @@ -90,10 +89,12 @@ private final class FdPollingSocket[F[_]: LiftIO] private ( def reads: Stream[F, Byte] = Stream.repeatEval(read(DefaultReadSize)).unNoneTerminate.unchunks - def write(bytes: Chunk[Byte]): F[Unit] = { + def write(bytes: Chunk[Byte]): F[Unit] = writeIncremental(bytes).compile.drain + + override def writeIncremental(bytes: Chunk[Byte]): Stream[F, Int] = { val Chunk.ArraySlice(buf, offset, length) = bytes.toArraySlice - def go(pos: Int): IO[Either[Int, Unit]] = + def writeNext(pos: Int): IO[Either[Int, Int]] = IO { if (LinktimeInfo.isLinux) guardSSize( @@ -101,18 +102,26 @@ private final class FdPollingSocket[F[_]: LiftIO] private ( ).toInt else guard(unistd.write(fd, buf.atUnsafe(offset + pos), (length - pos).toUSize)) - }.flatMap { wrote => - if (wrote >= 0) { + }.map { wrote => + if (wrote >= 0) + Right(wrote) + else + Left(pos) + } + + def writeAll(pos: Int): Pull[F, Int, Unit] = Pull + .eval(handle.pollWriteRec(pos)(writeNext(_)).to) + .flatMap { wrote => + Pull.output1(wrote) >> { val newPos = pos + wrote if (newPos < length) - go(newPos) + writeAll(newPos) else - IO.pure(Either.unit) - } else - IO.pure(Left(pos)) + Pull.done + } } - handle.pollWriteRec(0)(go(_)).to + writeAll(0).stream } def writes: Pipe[F, Byte, Nothing] = _.chunks.foreach(write(_)) diff --git a/io/native/src/main/scala/fs2/io/net/tls/S2nConnection.scala b/io/native/src/main/scala/fs2/io/net/tls/S2nConnection.scala index 3a26097f7d..557e2e8ffb 100644 --- a/io/native/src/main/scala/fs2/io/net/tls/S2nConnection.scala +++ b/io/native/src/main/scala/fs2/io/net/tls/S2nConnection.scala @@ -47,7 +47,7 @@ private[tls] trait S2nConnection[F[_]] { def read(n: Int): F[Option[Chunk[Byte]]] - def write(bytes: Chunk[Byte]): F[Unit] + def write(bytes: Chunk[Byte]): Stream[F, Int] def shutdown: F[Unit] @@ -164,22 +164,27 @@ private[tls] object S2nConnection { go(0) } - def write(bytes: Chunk[Byte]) = { + def write(bytes: Chunk[Byte]): Stream[F, Int] = { val Chunk.ArraySlice(buf, offset, n) = bytes.toArraySlice - def go(i: Int): F[Unit] = - F.delay { - writeTasks.set(F.unit) - val blocked = stackalloc[s2n_blocked_status]() - val wrote = guard(s2n_send(conn, buf.atUnsafe(offset + i), (n - i).toCSSize, blocked)) - (!blocked, Math.max(wrote, 0)) - }.productL(F.delay(writeTasks.get).flatten) + def go(i: Int): Pull[F, Int, Unit] = + Pull + .eval { + F.delay { + writeTasks.set(F.unit) + val blocked = stackalloc[s2n_blocked_status]() + val wrote = + guard(s2n_send(conn, buf.atUnsafe(offset + i), (n - i).toCSSize, blocked)) + (!blocked, Math.max(wrote, 0)) + }.productL(F.delay(writeTasks.get).flatten) + } .flatMap { case (blocked, wrote) => val total = i + wrote - go(total).unlessA(blocked.toInt == S2N_NOT_BLOCKED && total >= n) + Pull + .output1(wrote) >> go(total).unlessA(blocked.toInt == S2N_NOT_BLOCKED && total >= n) } - go(0) + go(0).stream } def shutdown = diff --git a/io/native/src/main/scala/fs2/io/net/tls/TLSSocketPlatform.scala b/io/native/src/main/scala/fs2/io/net/tls/TLSSocketPlatform.scala index 0cd7a7610f..8e9777d208 100644 --- a/io/native/src/main/scala/fs2/io/net/tls/TLSSocketPlatform.scala +++ b/io/native/src/main/scala/fs2/io/net/tls/TLSSocketPlatform.scala @@ -51,7 +51,10 @@ private[tls] trait TLSSocketCompanionPlatform { self: TLSSocket.type => writeMutex <- Mutex[F] } yield new UnsealedTLSSocket[F] { def write(bytes: Chunk[Byte]): F[Unit] = - writeMutex.lock.surround(connection.write(bytes)) + writeMutex.lock.surround(connection.write(bytes).compile.drain) + + override def writeIncremental(bytes: Chunk[Byte]): Stream[F, Int] = + Stream.resource(writeMutex.lock) >> connection.write(bytes) private def read0(maxBytes: Int): F[Option[Chunk[Byte]]] = connection.read(maxBytes) diff --git a/io/shared/src/main/scala/fs2/io/net/Socket.scala b/io/shared/src/main/scala/fs2/io/net/Socket.scala index 0751bd94db..2248695427 100644 --- a/io/shared/src/main/scala/fs2/io/net/Socket.scala +++ b/io/shared/src/main/scala/fs2/io/net/Socket.scala @@ -66,6 +66,22 @@ trait Socket[F[_]] extends SocketInfo[F] { */ def writes: Pipe[F, Byte, Nothing] + /** Writes `bytes` to the peer, reporting as bytes are written. + * + * Completes when the bytes are written to the socket. + * + * @note This default implementation will just `write` then report the total + * written. When possible, socket implementations in FS2 provide an override + * to provide more fine-grained reporting. + */ + def writeIncremental( + bytes: Chunk[Byte] + ): Stream[F, Int] = Stream.eval(write(bytes)) >> Stream(bytes.size) + + /** Writes the supplied stream of bytes to this socket via `writeIncremental` semantics. + */ + def writesIncremental: Pipe[F, Byte, Int] = s => s.chunks.flatMap(writeIncremental) + /** Reads a file and writes it to a socket. * Streams the file contents of the specified size and sends them over the socket. * The stream terminates when the entire file has reached end of file or the specified count is reached. diff --git a/io/shared/src/test/scala/fs2/io/net/SocketSuite.scala b/io/shared/src/test/scala/fs2/io/net/SocketSuite.scala index fe2e7c9e69..0c8a2a9114 100644 --- a/io/shared/src/test/scala/fs2/io/net/SocketSuite.scala +++ b/io/shared/src/test/scala/fs2/io/net/SocketSuite.scala @@ -134,6 +134,25 @@ class SocketSuite extends Fs2Suite with SocketSuitePlatform { .drain } + test("writeIncremental - records total bytes sent") { + val message = Chunk.array(("123456789012345678901234567890" * 10000).getBytes) + + Stream + .resource(setup) + .flatMap { case (server, clients) => + val readOnlyServer = server.map(_.reads).parJoinUnbounded + val client = + clients.take(1).flatMap { socket => + socket.writeIncremental(message) + } + + client.concurrently(readOnlyServer) + } + .compile + .foldMonoid + .map(bytes => assertEquals(bytes, message.size)) + } + test("addresses - should match across client and server sockets") { Stream .resource(setup)