Merge pull request #3551 from rahulrangers/issue-3171

Implement `Socket#sendFile`

Author: armanbilge

io/jvm-native/src/main/scala/fs2/io/file/FileHandlePlatform.scala

@@ -67,6 +67,47 @@ private[file] trait FileHandlePlatform[F[_]] {
 
 }
 
+private[io] final class SyncFileHandle[F[_]](val chan: FileChannel)(implicit F: Sync[F])
+    extends FileHandle[F] {
+  type Lock = FileLock
+
+  override def force(metaData: Boolean): F[Unit] =
+    F.blocking(chan.force(metaData))
+
+  override def lock: F[Lock] =
+    F.blocking(chan.lock)
+
+  override def lock(position: Long, size: Long, shared: Boolean): F[Lock] =
+    F.blocking(chan.lock(position, size, shared))
+
+  override def read(numBytes: Int, offset: Long): F[Option[Chunk[Byte]]] =
+    F.blocking {
+      val buf = ByteBuffer.allocate(numBytes)
+      val len = chan.read(buf, offset)
+      if (len < 0) None
+      else if (len == 0) Some(Chunk.empty)
+      else Some(Chunk.array(buf.array, 0, len))
+    }
+
+  override def size: F[Long] =
+    F.blocking(chan.size)
+
+  override def truncate(size: Long): F[Unit] =
+    F.blocking { chan.truncate(size); () }
+
+  override def tryLock: F[Option[Lock]] =
+    F.blocking(Option(chan.tryLock()))
+
+  override def tryLock(position: Long, size: Long, shared: Boolean): F[Option[Lock]] =
+    F.blocking(Option(chan.tryLock(position, size, shared)))
+
+  override def unlock(f: Lock): F[Unit] =
+    F.blocking(f.release())
+
+  override def write(bytes: Chunk[Byte], offset: Long): F[Int] =
+    F.blocking(chan.write(bytes.toByteBuffer, offset))
+}
+
 private[file] trait FileHandleCompanionPlatform {
   @deprecated("Use Files[F].open", "3.0.0")
   def fromPath[F[_]: Async](path: JPath, flags: Seq[OpenOption]): Resource[F, FileHandle[F]] =
@@ -76,49 +117,11 @@ private[file] trait FileHandleCompanionPlatform {
   def fromFileChannel[F[_]: Async](channel: F[FileChannel]): Resource[F, FileHandle[F]] =
     Files.forAsync[F].openFileChannel(channel)
 
-  /** Creates a `FileHandle[F]` from a `java.nio.channels.FileChannel`. */
+  /** Creates a `SyncFileHandle[F]` from a `java.nio.channels.FileChannel`. */
   private[file] def make[F[_]](
       chan: FileChannel
   )(implicit F: Sync[F]): FileHandle[F] =
-    new FileHandle[F] {
-      type Lock = FileLock
-
-      override def force(metaData: Boolean): F[Unit] =
-        F.blocking(chan.force(metaData))
-
-      override def lock: F[Lock] =
-        F.blocking(chan.lock)
-
-      override def lock(position: Long, size: Long, shared: Boolean): F[Lock] =
-        F.blocking(chan.lock(position, size, shared))
-
-      override def read(numBytes: Int, offset: Long): F[Option[Chunk[Byte]]] =
-        F.blocking {
-          val buf = ByteBuffer.allocate(numBytes)
-          val len = chan.read(buf, offset)
-          if (len < 0) None
-          else if (len == 0) Some(Chunk.empty)
-          else Some(Chunk.array(buf.array, 0, len))
-        }
-
-      override def size: F[Long] =
-        F.blocking(chan.size)
-
-      override def truncate(size: Long): F[Unit] =
-        F.blocking { chan.truncate(size); () }
-
-      override def tryLock: F[Option[Lock]] =
-        F.blocking(Option(chan.tryLock()))
-
-      override def tryLock(position: Long, size: Long, shared: Boolean): F[Option[Lock]] =
-        F.blocking(Option(chan.tryLock(position, size, shared)))
-
-      override def unlock(f: Lock): F[Unit] =
-        F.blocking(f.release())
-
-      override def write(bytes: Chunk[Byte], offset: Long): F[Int] =
-        F.blocking(chan.write(bytes.toByteBuffer, offset))
-    }
+    new SyncFileHandle[F](chan)
 
   /** Creates a `FileHandle[F]` from a `java.nio.channels.SeekableByteChannel`. Because a `SeekableByteChannel` doesn't provide all the functionalities required by `FileHandle` some features like locking will be unavailable. */
   private[file] def makeFromSeekableByteChannel[F[_]](

io/jvm/src/main/scala/fs2/io/net/SelectingSocket.scala

@@ -22,6 +22,8 @@
 package fs2
 package io.net
 
+import fs2.io.file.SyncFileHandle
+import fs2.io.file.FileHandle
 import cats.effect.LiftIO
 import cats.effect.Selector
 import cats.effect.kernel.Async
@@ -77,7 +79,32 @@ private final class SelectingSocket[F[_]: LiftIO] private (
     F.delay {
       ch.shutdownInput(); ()
     }
+  override def sendFile(
+      file: FileHandle[F],
+      offset: Long,
+      count: Long,
+      chunkSize: Int
+  ): Stream[F, Nothing] = file match {
+    case syncFileHandle: SyncFileHandle[F] =>
+      val fileChannel = syncFileHandle.chan
 
+      def go(currOffset: Long, remaining: Long): F[Unit] =
+        if (remaining <= 0) F.unit
+        else {
+          F.blocking(fileChannel.transferTo(currOffset, remaining, ch)).flatMap { written =>
+            if (written > 0) {
+              go(currOffset + written, remaining - written)
+            } else {
+              selector.select(ch, OP_WRITE).to *> go(currOffset, remaining)
+            }
+          }
+        }
+
+      Stream.exec(writeMutex.lock.surround(go(offset, count)))
+
+    case _ =>
+      super.sendFile(file, offset, count, chunkSize)
+  }
 }
 
 private object SelectingSocket {

io/jvm/src/main/scala/fs2/io/net/unixsocket/UnixSocketsPlatform.scala

@@ -20,7 +20,7 @@
  */
 
 package fs2.io.net.unixsocket
-
+import fs2.io.file.SyncFileHandle
 import cats.effect.IO
 import cats.effect.LiftIO
 import cats.effect.kernel.{Async, Resource}
@@ -34,6 +34,7 @@ import fs2.io.net.Socket
 import fs2.io.evalOnVirtualThreadIfAvailable
 import java.nio.ByteBuffer
 import java.nio.channels.SocketChannel
+import fs2.io.file.FileHandle
 
 private[unixsocket] trait UnixSocketsCompanionPlatform {
   def forIO: UnixSockets[IO] = forLiftIO
@@ -134,5 +135,30 @@ private[unixsocket] trait UnixSocketsCompanionPlatform {
           ch.shutdownInput(); ()
         }
       )
+    override def sendFile(
+        file: FileHandle[F],
+        offset: Long,
+        count: Long,
+        chunkSize: Int
+    ): Stream[F, Nothing] = file match {
+      case syncFileHandle: SyncFileHandle[F] =>
+        val fileChannel = syncFileHandle.chan
+
+        def go(currOffset: Long, remaining: Long): F[Unit] =
+          if (remaining <= 0) F.unit
+          else {
+            F.blocking(fileChannel.transferTo(currOffset, remaining, ch)).flatMap { written =>
+              if (written == 0) F.unit
+              else {
+                go(currOffset + written, remaining - written)
+              }
+            }
+          }
+
+        Stream.exec(writeMutex.lock.surround(go(offset, count)))
+
+      case _ =>
+        super.sendFile(file, offset, count, chunkSize)
+    }
   }
 }

io/shared/src/main/scala/fs2/io/net/Socket.scala

@@ -24,6 +24,7 @@ package io
 package net
 
 import com.comcast.ip4s.{IpAddress, SocketAddress}
+import fs2.io.file.FileHandle
 
 /** Provides the ability to read/write from a TCP socket in the effect `F`.
   */
@@ -69,6 +70,34 @@ trait Socket[F[_]] {
   /** Writes the supplied stream of bytes to this socket via `write` semantics.
     */
   def writes: Pipe[F, Byte, Nothing]
+
+  /** 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.
+    *
+    * @param file the file handle to read from
+    * @param offset the starting position in the file
+    * @param count the maximum number of bytes to transfer
+    * @param chunkSize the size of each chunk to read
+    */
+  def sendFile(
+      file: FileHandle[F],
+      offset: Long,
+      count: Long,
+      chunkSize: Int
+  ): Stream[F, Nothing] = {
+
+    def go(currOffset: Long, remaining: Long): Stream[F, Byte] =
+      if (remaining > 0)
+        Stream.eval(file.read(math.min(remaining, chunkSize.toLong).toInt, currOffset)).flatMap {
+          case Some(chunk) if chunk.nonEmpty =>
+            Stream.chunk(chunk) ++ go(currOffset + chunk.size, remaining - chunk.size)
+          case _ => Stream.empty
+        }
+      else Stream.empty
+
+    go(offset, count).through(writes)
+  }
 }
 
 object Socket extends SocketCompanionPlatform

io/shared/src/test/scala/fs2/io/net/tcp/SocketSuite.scala

@@ -29,6 +29,7 @@ import com.comcast.ip4s._
 
 import scala.concurrent.duration._
 import scala.concurrent.TimeoutException
+import fs2.io.file._
 
 class SocketSuite extends Fs2Suite with SocketSuitePlatform {
 
@@ -284,5 +285,45 @@ class SocketSuite extends Fs2Suite with SocketSuitePlatform {
         }
       }
     }
+    test("sendFile - sends data from file to socket from the offset") {
+      val content = "Hello, world!"
+      val offset = 7L
+      val count = 5L
+      val expected = "world"
+      val chunkSize = 2
+
+      val setup = for {
+        tempFile <- Files[IO].tempFile
+        _ <- Stream
+          .emits(content.getBytes)
+          .covary[IO]
+          .through(Files[IO].writeAll(tempFile))
+          .compile
+          .drain
+          .toResource
+        serverSetup <- Network[IO].serverResource(Some(ip"127.0.0.1"))
+        (bindAddress, server) = serverSetup
+        client <- Network[IO].client(bindAddress)
+      } yield (tempFile, server, client)
+
+      Stream
+        .resource(setup)
+        .flatMap { case (tempFile, server, client) =>
+          val serverStream = server.head.flatMap { socket =>
+            Stream.eval(socket.reads.through(text.utf8.decode).compile.string)
+          }
+          val clientStream =
+            Stream.resource(Files[IO].open(tempFile, Flags.Read)).flatMap { fileHandle =>
+              client.sendFile(fileHandle, offset, count, chunkSize).drain ++
+                Stream.eval(client.endOfOutput)
+            }
+          serverStream.concurrently(clientStream)
+        }
+        .compile
+        .lastOrError
+        .map { received =>
+          assertEquals(received, expected)
+        }
+    }
   }
 }