From 218bdfbfdd21483a8fae31395b6b0135dda3d867 Mon Sep 17 00:00:00 2001 From: Vibe Nuage Agent Date: Sat, 10 Oct 2026 15:57:18 +0000 Subject: [PATCH 1/3] Migrate HTTP layer from http4s to zio-http Serve routes, range requests, and the WebSocket session with zio-http 3.11.2, converting at the HTTP boundary via zio-interop-catz while keeping the cats-effect core unchanged. Co-authored-by: lavrov --- build.mill | 21 +- server/src/main/scala/Main.scala | 334 +++++++++++----------- server/src/main/scala/ServerTorrent.scala | 1 - server/src/main/scala/SocketSession.scala | 89 +++--- 4 files changed, 230 insertions(+), 215 deletions(-) diff --git a/build.mill b/build.mill index 3301848..0703b9f 100644 --- a/build.mill +++ b/build.mill @@ -17,10 +17,9 @@ object server extends CommonSettings { Deps.bittorrent.dht, Deps.`cats-effect`, Deps.`fs2-io`, - Deps.http4s.core, - Deps.http4s.dsl, - Deps.http4s.server, - Deps.http4s.client, + Deps.`zio-http`, + Deps.`zio-interop-cats`, + Deps.`ip4s-core`, Deps.`woof-core`, Deps.`woof-slf4j`, Deps.`cats-effect-cps`, @@ -59,7 +58,10 @@ object Deps { val fs2 = "3.12.2" val `scodec-bits` = "1.2.4" val upickle = "2.0.0" - val http4s = "1.0.0-M40" + val `zio-http` = "3.11.2" + val zio = "2.1.26" + val `zio-interop-cats` = "23.1.0.14" + val `ip4s-core` = "3.8.0" val `cats-effect-cps` = "0.9.21" val woof = "0.9.0" } @@ -73,12 +75,9 @@ object Deps { val `fs2-io` = mvn"co.fs2::fs2-io::${Versions.fs2}" val `scodec-bits` = mvn"org.scodec::scodec-bits::${Versions.`scodec-bits`}" val upickle = mvn"com.lihaoyi::upickle::${Versions.upickle}" - object http4s { - val core = mvn"org.http4s::http4s-core::${Versions.http4s}" - val dsl = mvn"org.http4s::http4s-dsl::${Versions.http4s}" - val server = mvn"org.http4s::http4s-ember-server::${Versions.http4s}" - val client = mvn"org.http4s::http4s-ember-client::${Versions.http4s}" - } + val `zio-http` = mvn"dev.zio::zio-http::${Versions.`zio-http`}" + val `zio-interop-cats` = mvn"dev.zio::zio-interop-cats::${Versions.`zio-interop-cats`}" + val `ip4s-core` = mvn"com.comcast::ip4s-core::${Versions.`ip4s-core`}" val `cats-effect-cps` = mvn"com.github.rssh::cps-async-connect-cats-effect::${Versions.`cats-effect-cps`}" val `woof-core` = mvn"org.legogroup::woof-core::${Versions.woof}" val `woof-slf4j` = mvn"org.legogroup::woof-slf4j::${Versions.woof}" diff --git a/server/src/main/scala/Main.scala b/server/src/main/scala/Main.scala index eb35904..292c451 100644 --- a/server/src/main/scala/Main.scala +++ b/server/src/main/scala/Main.scala @@ -1,28 +1,27 @@ -import cats.syntax.all.* -import cats.effect.syntax.all.* -import cats.data.{Kleisli, OptionT} +import cats.arrow.FunctionK +import cats.data.OptionT import cats.effect.{ExitCode, IO, IOApp, Resource, ResourceIO} import cats.effect.std.{Dispatcher, Random} -import com.github.torrentdam.bittorrent.dht.{Node, NodeId, PeerDiscovery, QueryHandler, RoutingTable, RoutingTableBootstrap} -import com.github.torrentdam.bittorrent.wire.{Connection, Swarm} -import com.github.torrentdam.bittorrent.{FileMapping, InfoHash, PeerId, TorrentFile} +import cats.effect.syntax.all.* +import cats.syntax.all.* +import com.comcast.ip4s.{Host, SocketAddress} import com.github.torrentdam.bencode.encode import com.github.torrentdam.bencode.format.BencodeFormat +import com.github.torrentdam.bittorrent.InfoHash +import com.github.torrentdam.bittorrent.dht.{Node, NodeId, PeerDiscovery, QueryHandler, RoutingTable, RoutingTableBootstrap} +import com.github.torrentdam.bittorrent.wire.Connection +import com.github.torrentdam.bittorrent.{FileMapping, PeerId, TorrentFile} import fs2.Stream -import fs2.io.net.{DatagramSocketGroup, Network, SocketGroup} -import org.http4s.headers.{Range, `Accept-Ranges`, `Content-Disposition`, `Content-Length`, `Content-Range`, `Content-Type`} -import org.http4s.ember.server.EmberServerBuilder -import org.http4s.{HttpApp, MediaType, Response} import org.legogroup.woof.{Logger, given} import org.legogroup.woof.slf4j.registerSlf4j -import org.typelevel.log4cats.LoggerFactory -import org.typelevel.log4cats.slf4j.Slf4jFactory -import org.typelevel.ci.* import sun.misc.Signal -import Routes.FileIndex -import com.comcast.ip4s.{Host, SocketAddress} -import org.http4s.ember.client.EmberClientBuilder -import org.http4s.server.websocket.WebSocketBuilder +import zio.http.* +import zio.http.codec.PathCodec.{int, string} +import zio.interop.fromEffect +import zio.interop.ToEffectSyntax +import zio.interop.catz.implicits.rts +import zio.stream.interop.fs2z.* +import zio.{Chunk, RIO, ZIO} import cps.* import cps.syntax.* import cps.monads.catsEffect.{*, given} @@ -46,11 +45,10 @@ object Main extends IOApp { given logger: Logger[IO] = !mkLogger given Dispatcher[IO] = !Dispatcher.sequential[IO] !logger.registerSlf4j - given LoggerFactory[IO] = Slf4jFactory.create[IO] !registerSignalHandler - val app: HttpWebSocketApp = !makeApp(dhtBootstrapNode) + val routes = !makeApp(dhtBootstrapNode) val bindPort = Option(System.getenv("PORT")).flatMap(_.toIntOption).getOrElse(9999) - !serve(bindPort, app) + !serve(bindPort, routes) ExitCode.Success } @@ -83,141 +81,177 @@ object Main extends IOApp { (torrentRegistry, createServerTorrent, torrentIndex, metadataRegistry) } + type FileIndex = Int - type HttpWebSocketApp = WebSocketBuilder[IO] => HttpApp[IO] + def ioToRIO(using Dispatcher[IO]): FunctionK[IO, RIO[Any, _]] = + new FunctionK[IO, RIO[Any, _]] { + def apply[A](fa: IO[A]): RIO[Any, A] = fromEffect(fa) + } def makeApp( dhtBootstrapNode: Option[SocketAddress[Host]] - )(using Logger[IO]): Resource[IO, HttpWebSocketApp] = async[Resource[IO, _]]{ - import org.http4s.dsl.io.* - val (torrentRegistry, createServerTorrent, torrentIndex, metadataRegistry) = !resources(dhtBootstrapNode) - - def handleSocket(wsBuilder: WebSocketBuilder[IO]) = + )(using Logger[IO], Dispatcher[IO]): Resource[IO, Routes[Any, Response]] = + resources(dhtBootstrapNode).map { (torrentRegistry, createServerTorrent, torrentIndex, metadataRegistry) => val makeTorrent: SocketSession.MakeTorrent = (infoHash, trackers) => torrentRegistry.getOrCreate(infoHash)(createServerTorrent(infoHash, trackers)) - SocketSession(makeTorrent, metadataRegistry, torrentIndex, wsBuilder) - def handleGetTorrent(infoHash: InfoHash) = - torrentRegistry - .get(infoHash) - .use { torrent => - val metadata = torrent.metadata - val torrentFile = TorrentFile(metadata, None) - val bcode = - summon[BencodeFormat[TorrentFile]] - .write(torrentFile) - .toOption - .get - val filename = metadata.parsed.files match { - case file :: Nil if file.path.nonEmpty => file.path.last - case _ => infoHash.bytes.toHex - } - val bytes = encode(bcode) - OptionT.liftF( - Ok( - bytes.toByteArray, - `Content-Disposition`("inline", Map(ci"filename" -> s"$filename.torrent")) - ) - ) + val handleSocket = SocketSession(makeTorrent, metadataRegistry, torrentIndex) + + def handleGetTorrent(infoHash: InfoHash): ZIO[Any, Nothing, Response] = + fromEffect { + torrentRegistry + .get(infoHash) + .use { torrent => + val metadata = torrent.metadata + val torrentFile = TorrentFile(metadata, None) + val bcode = + summon[BencodeFormat[TorrentFile]] + .write(torrentFile) + .toOption + .get + val filename = metadata.parsed.files match { + case file :: Nil if file.path.nonEmpty => file.path.last + case _ => infoHash.bytes.toHex + } + OptionT.liftF(IO.pure((s"$filename.torrent", encode(bcode).toByteArray))) + } + .value } - .getOrElseF { - NotFound("Torrent not found") + .map { + case Some((filename, bytes)) => + Response( + status = Status.Ok, + headers = Headers( + Header.Custom("content-disposition", s"""inline; filename="$filename"""") + ), + body = Body.fromChunk(Chunk.fromArray(bytes)), + ) + case None => + Response.error(Status.NotFound, "Torrent not found") } + .catchAll(_ => ZIO.succeed(Response.status(Status.InternalServerError))) - def handleGetData(infoHash: InfoHash, fileIndex: FileIndex, rangeOpt: Option[Range]) = - torrentRegistry.get(infoHash).allocated.value.flatMap { - case Some((torrent: ServerTorrent, release)) => - if torrent.files.value.lift(fileIndex).isDefined then { - val file = torrent.metadata.parsed.files(fileIndex) - val extension = file.path.lastOption.map(_.reverse.takeWhile(_ != '.').reverse) - val fileMapping = torrent.files - val parallelPieces = scala.math.max(maxPrefetchBytes / torrent.metadata.parsed.pieceLength, 2).toInt - def downloadPiece(index: Long) = - torrent - .piece(index.toInt) - .timeoutTo( - downloadPieceTimeout, - IO.raiseError(PieceDownloadTimeout(index)) - ) - .tupleLeft(index) - def dataStream(span: FileMapping.FileSpan) = - ( - Stream - .emits(span.beginIndex to span.endIndex) - .covary[IO] - .parEvalMap(parallelPieces)(downloadPiece) - ) - .flatMap { - case (span.beginIndex, bytes) => - bytes.drop(span.beginOffset) - case (span.endIndex, bytes) => - bytes.take(span.endOffset) - case (_, bytes) => bytes - } - .onFinalize(release.value.void) + def handleGetData( + infoHash: InfoHash, + fileIndex: FileIndex, + rangeOpt: Option[Header.Range], + ): ZIO[Any, Nothing, Response] = + fromEffect { torrentRegistry.get(infoHash).allocated.value } + .catchAll(_ => ZIO.succeed(None: Option[(ServerTorrent, TorrentRegistry.Optional[Unit])])) + .flatMap { + case Some((torrent: ServerTorrent, release)) => + if torrent.files.value.lift(fileIndex).isDefined then + ZIO.succeed { + val file = torrent.metadata.parsed.files(fileIndex) + val extension = file.path.lastOption.map(_.reverse.takeWhile(_ != '.').reverse) + val fileMapping = torrent.files + val parallelPieces = scala.math.max(maxPrefetchBytes / torrent.metadata.parsed.pieceLength, 2).toInt + def downloadPiece(index: Long) = + torrent + .piece(index.toInt) + .timeoutTo( + downloadPieceTimeout, + IO.raiseError(PieceDownloadTimeout(index)) + ) + .tupleLeft(index) + def dataStream(span: FileMapping.FileSpan) = + ( + Stream + .emits(span.beginIndex to span.endIndex) + .covary[IO] + .parEvalMap(parallelPieces)(downloadPiece) + ) + .flatMap { + case (span.beginIndex, bytes) => + bytes.drop(span.beginOffset) + case (span.endIndex, bytes) => + bytes.take(span.endOffset) + case (_, bytes) => bytes + } + .onFinalize(release.value.void) - val mediaType = - extension.flatMap(MediaType.forExtension).getOrElse(MediaType.application.`octet-stream`) - val span0 = fileMapping.value(fileIndex) - rangeOpt match { - case Some(range) => - val first = range.ranges.head.first - val second = range.ranges.head.second - val advanced = span0.advance(fileMapping.pieceLength, first) - val span = second.fold(advanced) { second => - advanced.take(fileMapping.pieceLength, second - first) - } - val subRange = rangeOpt match { - case Some(range) => - val first = range.ranges.head.first - val second = range.ranges.head.second.getOrElse(file.length - 1) - Range.SubRange(first, second) - case None => - Range.SubRange(0L, file.length - 1) + def responseBody(stream: Stream[IO, Byte], contentLength: Long) = + Body.fromStream(stream.translate(ioToRIO).toZStream(), contentLength) + + val mediaType = + extension + .flatMap(MediaType.forFileExtension) + .getOrElse(MediaType.application.`octet-stream`) + val span0 = fileMapping.value(fileIndex) + val rangeBounds: Option[(Long, Option[Long])] = + rangeOpt.flatMap { + case Header.Range.Single("bytes", first, second) => + Some((first, second)) + case Header.Range.Prefix("bytes", first) => + Some((first, None)) + case _ => None + } + rangeBounds match + case Some((first, second)) => + val advanced = span0.advance(fileMapping.pieceLength, first) + val span = second.fold(advanced) { second => + advanced.take(fileMapping.pieceLength, second - first) + } + val subRangeFirst = first + val subRangeLast = second.getOrElse(file.length - 1) + Response( + status = Status.PartialContent, + headers = Headers( + Header.AcceptRanges.Bytes, + Header.ContentType(mediaType), + Header.Custom("content-range", s"bytes $subRangeFirst-$subRangeLast/${file.length}"), + ), + body = responseBody(dataStream(span), file.length - subRangeFirst), + ) + case None => + val filename = file.path.lastOption.getOrElse(s"file-$fileIndex") + Response( + status = Status.Ok, + headers = Headers( + Header.AcceptRanges.Bytes, + Header.ContentType(mediaType), + Header.Custom("content-disposition", s"""inline; filename="$filename""""), + ), + body = responseBody(dataStream(span0), file.length), + ) } - PartialContent( - dataStream(span), - `Content-Type`(mediaType), - `Accept-Ranges`.bytes, - `Content-Range`(subRange, file.length.some), - `Content-Length`.unsafeFromLong(file.length - subRange.first) - ) - case None => - val filename = file.path.lastOption.getOrElse(s"file-$fileIndex") - Ok( - dataStream(span0), - `Accept-Ranges`.bytes, - `Content-Type`(mediaType), - `Content-Disposition`("inline", Map(ci"filename" -> filename)), - `Content-Length`.unsafeFromLong(file.length) + else + ZIO.succeed( + Response.error(Status.NotFound, s"Torrent does not contain file with index $fileIndex") ) - } - } - else { - NotFound(s"Torrent does not contain file with index $fileIndex") + case None => + ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) } - case None => NotFound("Torrent not found") - } - webSocketBuilder => - Routes.routes( - handleSocket(webSocketBuilder), - handleGetTorrent, - handleGetData + Routes( + Method.GET / "" -> handler(Response.text("Success")), + Method.GET / "ws" -> handler(handleSocket.toResponse), + Method.GET / "torrent" / string("infoHash") / "metadata" -> handler { + (infoHash: String, _: Request) => + InfoHash.fromString.lift(infoHash) match + case Some(infoHash) => + handleGetTorrent(infoHash) + case None => + ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) + }, + Method.GET / "torrent" / string("infoHash") / "data" / int("fileIndex") -> handler { + (infoHash: String, fileIndex: Int, req: Request) => + val validFileIndex = Option.when(fileIndex >= 0)(fileIndex) + (InfoHash.fromString.lift(infoHash), validFileIndex) match + case (Some(infoHash), Some(fileIndex)) => + handleGetData(infoHash, fileIndex, req.header(Header.Range)) + case _ => + ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) + }, ) - } + } - def serve(bindPort: Int, app: HttpWebSocketApp)(using LoggerFactory[IO]): IO[Nothing] = - import com.comcast.ip4s.{host, Port} - EmberServerBuilder - .default[IO] - .withHttpWebSocketApp(app) - .withHost(host"0.0.0.0") - .withPort(Port.fromInt(bindPort).get) - .withHttp2 - .build - .useForever + def serve(bindPort: Int, routes: Routes[Any, Response])(using Dispatcher[IO]): IO[Nothing] = + Server + .serve(routes) + .provide(Server.defaultWithPort(bindPort)) + .toEffect[IO] case class PieceDownloadTimeout(index: Long) extends Throwable(s"Timeout downloading piece $index") @@ -242,29 +276,3 @@ object Main extends IOApp { self.copy(endIndex = s0.beginIndex, endOffset = self.beginOffset) } } - -object Routes { - val dsl = org.http4s.dsl.io - - def routes( - handleSocket: IO[Response[IO]], - handleGetTorrent: InfoHash => IO[Response[IO]], - handleGetData: (InfoHash, FileIndex, Option[Range]) => IO[Response[IO]], - ): HttpApp[IO] = { - import dsl.* - Kleisli { - case GET -> Root => Ok("Success") - case GET -> Root / "ws" => handleSocket - case GET -> Root / "torrent" / InfoHash.fromString(infoHash) / "metadata" => - handleGetTorrent(infoHash) - case req @ GET -> Root / "torrent" / InfoHash.fromString(infoHash) / "data" / FileIndexVar(index) => - handleGetData(infoHash, index, req.headers.get[Range]) - case _ => NotFound() - } - } - - type FileIndex = Int - val FileIndexVar: PartialFunction[String, FileIndex] = Function.unlift { (in: String) => - in.toIntOption.filter(_ >= 0) - } -} diff --git a/server/src/main/scala/ServerTorrent.scala b/server/src/main/scala/ServerTorrent.scala index 7008ea5..0d46558 100644 --- a/server/src/main/scala/ServerTorrent.scala +++ b/server/src/main/scala/ServerTorrent.scala @@ -9,7 +9,6 @@ import com.github.torrentdam.bittorrent.TorrentMetadata.Lossless import com.github.torrentdam.bittorrent.dht.PeerDiscovery import fs2.Stream import fs2.concurrent.Signal -import org.http4s.Uri import org.legogroup.woof.{Logger, given} trait ServerTorrent { diff --git a/server/src/main/scala/SocketSession.scala b/server/src/main/scala/SocketSession.scala index 8a2e56b..8492b0e 100644 --- a/server/src/main/scala/SocketSession.scala +++ b/server/src/main/scala/SocketSession.scala @@ -1,15 +1,14 @@ -import cats.effect.kernel.Deferred import cats.effect.{Concurrent, IO, Resource} -import cats.effect.std.Queue import cats.effect.std.Supervisor import cats.implicits.* import com.github.torrentdam.bittorrent.InfoHash import com.github.lavrov.bittorrent.app.protocol.{Message, Event, Command} -import fs2.Stream import org.legogroup.woof.{Logger, given} -import org.http4s.Response -import org.http4s.server.websocket.WebSocketBuilder -import org.http4s.websocket.WebSocketFrame +import zio.http.{ChannelEvent, Handler, WebSocketApp, WebSocketFrame} +import zio.interop.catz.implicits.rts +import zio.interop.ToEffectSyntax +import zio.interop.fromEffect +import zio.ZIO import cps.* import cps.syntax.* import cps.monads.catsEffect.{*, given} @@ -22,45 +21,55 @@ object SocketSession { def apply( makeTorrent: MakeTorrent, metadataRegistry: MetadataRegistry[IO], - torrentIndex: TorrentIndex, - webSocketBuilder: WebSocketBuilder[IO] + torrentIndex: TorrentIndex )(using F: Concurrent[IO], + dispatcher: cats.effect.std.Dispatcher[IO], logger: Logger[IO] - ): IO[Response[IO]] = - for - _ <- logger.info("Session started") - input <- Queue.unbounded[IO, WebSocketFrame] - output <- Queue.unbounded[IO, WebSocketFrame] - send = (str: String) => output.offer(WebSocketFrame.Text(str)) - sendMessage = (m: Message) => send(upickle.default.write(m)) - (handler, closeHandler) <- CommandHandler(sendMessage, makeTorrent, metadataRegistry, torrentIndex).allocated - fiber <- processor(input, sendMessage, handler).compile.drain.start - pingFiber <- (IO.sleep(10.seconds) >> input.offer(WebSocketFrame.Ping())).foreverM.start - response <- webSocketBuilder - .withOnClose(fiber.cancel >> pingFiber.cancel >> closeHandler >> logger.info("Session closed")) - .build( - Stream.fromQueueUnterminated(output), - _.evalMap(input.offer), - ) - yield response - - private def processor( - input: Queue[IO, WebSocketFrame], - send: Message => IO[Unit], - commandHandler: CommandHandler - )(using logger: Logger[IO]): Stream[IO, Unit] = - Stream.fromQueueUnterminated(input).evalMap { - case WebSocketFrame.Text(JsonMessage(message), _) => - message match + ): WebSocketApp[Any] = + Handler.webSocket { channel => + val send: Message => ZIO[Any, Nothing, Unit] = { message => + channel + .send(ChannelEvent.Read(WebSocketFrame.Text(upickle.default.write(message)))) + .catchAll(_ => ZIO.unit) + } + val sendEvent: Event => IO[Unit] = { event => + send(event).toEffect[IO] + } + val allocatedIO: IO[(CommandHandler, IO[Unit])] = + CommandHandler(sendEvent, makeTorrent, metadataRegistry, torrentIndex).allocated + fromEffect(logger.info("Session started")) *> fromEffect(allocatedIO).flatMap { allocated => + val handler = allocated._1 + val closeHandler = allocated._2 + def process: Message => ZIO[Any, Nothing, Unit] = { case Message.Ping => send(Message.Pong) - case command: Command => async[IO] { - !logger.debug(s"Received $command") - !commandHandler.handle(command) + case command: Command => + fromEffect { + async[IO] { + !logger.debug(s"Received $command") + !handler.handle(command) + } + }.catchAll(_ => ZIO.unit) + case _ => + ZIO.unit + } + val receiveLoop = channel.receiveAll { + case ChannelEvent.Read(WebSocketFrame.Text(text)) => + JsonMessage.lift(text).map(process).getOrElse(ZIO.unit) + case _ => + ZIO.unit + } + val pingLoop = ZIO.sleep(zio.Duration.fromScala(10.seconds)) *> + channel.send(ChannelEvent.Read(WebSocketFrame.Ping)).catchAll(_ => ZIO.unit) + receiveLoop + .zipPar(pingLoop.forever) + .ensuring { + fromEffect { + closeHandler *> logger.info("Session closed") + }.catchAll(_ => ZIO.unit) } - case _ => IO.unit - case _ => IO.unit + } } private val JsonMessage: PartialFunction[String, Message] = @@ -150,7 +159,7 @@ object SocketSession { supervisor ) } - + trait MakeTorrent { def apply(infoHash: InfoHash, trackers: List[String]): Resource[IO, IO[ServerTorrent.Phase.PeerDiscovery]] } From d233aa9448aa77fa2c7b802b8b73bc200b5b6bac Mon Sep 17 00:00:00 2001 From: Vibe Nuage Agent Date: Sat, 10 Oct 2026 21:45:59 +0000 Subject: [PATCH 2/3] Make entry point and HTTP layer ZIO-native Replace IOApp with ZIOAppDefault. Main.run is a ZIO for-comprehension that acquires a cats-effect Dispatcher via ZIO scope, serves zio-http routes directly, and converts to the cats-effect core only through fromEffect at the boundary. Co-authored-by: lavrov --- server/src/main/scala/Main.scala | 312 ++++++++++++++----------------- 1 file changed, 140 insertions(+), 172 deletions(-) diff --git a/server/src/main/scala/Main.scala b/server/src/main/scala/Main.scala index 292c451..3bfd23f 100644 --- a/server/src/main/scala/Main.scala +++ b/server/src/main/scala/Main.scala @@ -1,6 +1,5 @@ -import cats.arrow.FunctionK import cats.data.OptionT -import cats.effect.{ExitCode, IO, IOApp, Resource, ResourceIO} +import cats.effect.{IO, Resource} import cats.effect.std.{Dispatcher, Random} import cats.effect.syntax.all.* import cats.syntax.all.* @@ -8,7 +7,7 @@ import com.comcast.ip4s.{Host, SocketAddress} import com.github.torrentdam.bencode.encode import com.github.torrentdam.bencode.format.BencodeFormat import com.github.torrentdam.bittorrent.InfoHash -import com.github.torrentdam.bittorrent.dht.{Node, NodeId, PeerDiscovery, QueryHandler, RoutingTable, RoutingTableBootstrap} +import com.github.torrentdam.bittorrent.dht.Node import com.github.torrentdam.bittorrent.wire.Connection import com.github.torrentdam.bittorrent.{FileMapping, PeerId, TorrentFile} import fs2.Stream @@ -18,222 +17,200 @@ import sun.misc.Signal import zio.http.* import zio.http.codec.PathCodec.{int, string} import zio.interop.fromEffect -import zio.interop.ToEffectSyntax -import zio.interop.catz.implicits.rts import zio.stream.interop.fs2z.* -import zio.{Chunk, RIO, ZIO} +import zio.{Chunk, RIO, Runtime, Scope, ZIO, ZIOAppArgs, ZIOAppDefault, durationInt} import cps.* import cps.syntax.* import cps.monads.catsEffect.{*, given} import scala.concurrent.duration.* -object Main extends IOApp { +object Main extends ZIOAppDefault { def downloadPieceTimeout: FiniteDuration = 3.minutes def maxPrefetchBytes = 50 * 1000 * 1000 - def run(args: List[String]): IO[ExitCode] = asyncScope[IO]{ - val dhtBootstrapNode = args match { - case addressString :: _ => - val address = !SocketAddress - .fromString(addressString) - .liftTo[ResourceIO](new IllegalArgumentException("Invalid host")) - Some(address) - case _ => None - } - given logger: Logger[IO] = !mkLogger - given Dispatcher[IO] = !Dispatcher.sequential[IO] - !logger.registerSlf4j - !registerSignalHandler - val routes = !makeApp(dhtBootstrapNode) - val bindPort = Option(System.getenv("PORT")).flatMap(_.toIntOption).getOrElse(9999) - !serve(bindPort, routes) - ExitCode.Success - } - - def registerSignalHandler: IO[Unit] = - IO { + given bootstrapRuntime: Runtime[Any] = Runtime.default + + def run: ZIO[Environment & ZIOAppArgs & Scope, Any, Any] = + for + args <- getArgs.map(_.toList) + dispatcher <- acquireDispatcher + given Dispatcher[IO] = dispatcher + dhtBootstrapNode <- parseBootstrapNode(args.headOption) + logger <- fromEffect(mkLogger) + given Logger[IO] = logger + _ <- fromEffect(logger.registerSlf4j) + _ <- registerSignalHandler + bindPort = Option(System.getenv("PORT")).flatMap(_.toIntOption).getOrElse(9999) + routes <- makeRoutes(dhtBootstrapNode) + _ <- Server.serve(routes).provide(Server.defaultWithPort(bindPort)) + yield () + + def acquireDispatcher: ZIO[Scope, Nothing, Dispatcher[IO]] = + ZIO.acquireReleaseExit { + ZIO.succeed(Dispatcher.sequential[IO].allocated.unsafeRunSync()(using cats.effect.unsafe.IORuntime.global)) + } { case ((_, release), _) => + ZIO.succeed(release.unsafeRunAndForget()(using cats.effect.unsafe.IORuntime.global)) + }.map(_._1) + + def parseBootstrapNode(addressString: Option[String]): ZIO[Any, Throwable, Option[SocketAddress[Host]]] = + addressString match + case Some(value) => + ZIO + .fromOption(SocketAddress.fromString(value)) + .mapError(_ => new IllegalArgumentException("Invalid host")) + .map(Some(_)) + case None => + ZIO.succeed(None) + + def acquireCats[A](resource: Resource[IO, A])(using d: Dispatcher[IO] = null): ZIO[Scope, Throwable, A] = + for + allocated <- fromEffect(resource.allocated) + (value, release) = allocated + _ <- ZIO.addFinalizer(fromEffect(release).orDie) + yield value + + def registerSignalHandler: ZIO[Any, Nothing, Unit] = + ZIO.succeed { Signal.handle(new Signal("INT"), _ => System.exit(0)) } - // suppress stack trace report to circumvent https://github.com/typelevel/cats-effect/issues/3086 - override def reportFailure(err: Throwable): IO[Unit] = IO.unit - def resources( - dhtBootstrapNode: Option[SocketAddress[Host]] - )( - using Logger[IO] - ): Resource[IO, (TorrentRegistry, ServerTorrent.Create, TorrentIndex, MetadataRegistry[IO])] = - async[Resource[IO, _]]{ - given Random[IO] = !Resource.eval { Random.scalaUtilRandom[IO] } - val selfId = !Resource.eval { PeerId.generate } - val dhtNode = !Node(port = none, bootstrapNodeAddress = dhtBootstrapNode) - val peerDiscovery = dhtNode.discovery - val metadataRegistry = !Resource.eval { MetadataRegistry[IO]() } - val createServerTorrent = new ServerTorrent.Create( - infoHash => peerInfo => Connection.connect(selfId, peerInfo, infoHash), - peerDiscovery, - metadataRegistry - ) - val torrentRegistry = !Resource.eval { TorrentRegistry() } - val torrentIndex = !TorrentIndex() - (torrentRegistry, createServerTorrent, torrentIndex, metadataRegistry) - } + dhtBootstrapNode: Option[SocketAddress[Host]] + )(using Dispatcher[IO], Logger[IO]): Resource[IO, (TorrentRegistry, ServerTorrent.Create, TorrentIndex, MetadataRegistry[IO])] = + for + random <- Resource.eval(Random.scalaUtilRandom[IO]) + selfId <- Resource.eval(PeerId.generate) + dhtNode <- Node(port = none, bootstrapNodeAddress = dhtBootstrapNode) + metadataRegistry <- Resource.eval(MetadataRegistry[IO]()) + createServerTorrent = new ServerTorrent.Create( + infoHash => peerInfo => Connection.connect(selfId, peerInfo, infoHash), + dhtNode.discovery, + metadataRegistry + ) + torrentRegistry <- Resource.eval(TorrentRegistry()) + torrentIndex <- TorrentIndex() + yield (torrentRegistry, createServerTorrent, torrentIndex, metadataRegistry) type FileIndex = Int - def ioToRIO(using Dispatcher[IO]): FunctionK[IO, RIO[Any, _]] = - new FunctionK[IO, RIO[Any, _]] { + def ioToRIO(using Dispatcher[IO]): cats.arrow.FunctionK[IO, RIO[Any, _]] = + new cats.arrow.FunctionK[IO, RIO[Any, _]] { def apply[A](fa: IO[A]): RIO[Any, A] = fromEffect(fa) } - def makeApp( + def makeRoutes( dhtBootstrapNode: Option[SocketAddress[Host]] - )(using Logger[IO], Dispatcher[IO]): Resource[IO, Routes[Any, Response]] = - resources(dhtBootstrapNode).map { (torrentRegistry, createServerTorrent, torrentIndex, metadataRegistry) => - val makeTorrent: SocketSession.MakeTorrent = + )(using Dispatcher[IO], Logger[IO]): ZIO[Scope, Throwable, Routes[Any, Response]] = + for + (torrentRegistry, createServerTorrent, torrentIndex, metadataRegistry) <- acquireCats(resources(dhtBootstrapNode)) + makeTorrent: SocketSession.MakeTorrent = (infoHash, trackers) => torrentRegistry.getOrCreate(infoHash)(createServerTorrent(infoHash, trackers)) + handleSocket = SocketSession(makeTorrent, metadataRegistry, torrentIndex) - val handleSocket = SocketSession(makeTorrent, metadataRegistry, torrentIndex) - - def handleGetTorrent(infoHash: InfoHash): ZIO[Any, Nothing, Response] = + handleGetTorrent = (infoHash: InfoHash) => fromEffect { torrentRegistry .get(infoHash) .use { torrent => val metadata = torrent.metadata val torrentFile = TorrentFile(metadata, None) - val bcode = - summon[BencodeFormat[TorrentFile]] - .write(torrentFile) - .toOption - .get - val filename = metadata.parsed.files match { + val bcode = summon[BencodeFormat[TorrentFile]].write(torrentFile).toOption.get + val filename = metadata.parsed.files match case file :: Nil if file.path.nonEmpty => file.path.last case _ => infoHash.bytes.toHex - } OptionT.liftF(IO.pure((s"$filename.torrent", encode(bcode).toByteArray))) } .value - } - .map { + }.map { case Some((filename, bytes)) => Response( status = Status.Ok, - headers = Headers( - Header.Custom("content-disposition", s"""inline; filename="$filename"""") - ), + headers = Headers(Header.Custom("content-disposition", s"""inline; filename="$filename"""")), body = Body.fromChunk(Chunk.fromArray(bytes)), ) case None => Response.error(Status.NotFound, "Torrent not found") - } - .catchAll(_ => ZIO.succeed(Response.status(Status.InternalServerError))) + }.catchAll(_ => ZIO.succeed(Response.status(Status.InternalServerError))) - def handleGetData( - infoHash: InfoHash, - fileIndex: FileIndex, - rangeOpt: Option[Header.Range], - ): ZIO[Any, Nothing, Response] = - fromEffect { torrentRegistry.get(infoHash).allocated.value } + handleGetData = (infoHash: InfoHash, fileIndex: FileIndex, rangeOpt: Option[Header.Range]) => + fromEffect(torrentRegistry.get(infoHash).allocated.value) .catchAll(_ => ZIO.succeed(None: Option[(ServerTorrent, TorrentRegistry.Optional[Unit])])) .flatMap { case Some((torrent: ServerTorrent, release)) => - if torrent.files.value.lift(fileIndex).isDefined then - ZIO.succeed { - val file = torrent.metadata.parsed.files(fileIndex) - val extension = file.path.lastOption.map(_.reverse.takeWhile(_ != '.').reverse) - val fileMapping = torrent.files - val parallelPieces = scala.math.max(maxPrefetchBytes / torrent.metadata.parsed.pieceLength, 2).toInt - def downloadPiece(index: Long) = - torrent - .piece(index.toInt) - .timeoutTo( - downloadPieceTimeout, - IO.raiseError(PieceDownloadTimeout(index)) - ) - .tupleLeft(index) - def dataStream(span: FileMapping.FileSpan) = - ( - Stream - .emits(span.beginIndex to span.endIndex) - .covary[IO] - .parEvalMap(parallelPieces)(downloadPiece) - ) + ZIO.succeed { + val file = torrent.metadata.parsed.files(fileIndex) + val extension = file.path.lastOption.map(_.reverse.takeWhile(_ != '.').reverse) + val fileMapping = torrent.files + val parallelPieces = scala.math.max(maxPrefetchBytes / torrent.metadata.parsed.pieceLength, 2).toInt + def downloadPiece(index: Long): IO[(Long, Stream[IO, Byte])] = + torrent + .piece(index.toInt) + .timeoutTo(downloadPieceTimeout, IO.raiseError(PieceDownloadTimeout(index))) + .tupleLeft(index) + def dataStream(span: FileMapping.FileSpan): Stream[IO, Byte] = + Stream + .emits(span.beginIndex to span.endIndex) + .covary[IO] + .parEvalMap(parallelPieces)(downloadPiece) .flatMap { - case (span.beginIndex, bytes) => - bytes.drop(span.beginOffset) - case (span.endIndex, bytes) => - bytes.take(span.endOffset) - case (_, bytes) => bytes + case (beginIndex, bytes) if beginIndex == span.beginIndex => bytes.drop(span.beginOffset) + case (endIndex, bytes) if endIndex == span.endIndex => bytes.take(span.endOffset) + case (_, bytes) => bytes } .onFinalize(release.value.void) - - def responseBody(stream: Stream[IO, Byte], contentLength: Long) = - Body.fromStream(stream.translate(ioToRIO).toZStream(), contentLength) - - val mediaType = - extension - .flatMap(MediaType.forFileExtension) - .getOrElse(MediaType.application.`octet-stream`) - val span0 = fileMapping.value(fileIndex) - val rangeBounds: Option[(Long, Option[Long])] = - rangeOpt.flatMap { - case Header.Range.Single("bytes", first, second) => - Some((first, second)) - case Header.Range.Prefix("bytes", first) => - Some((first, None)) - case _ => None - } - rangeBounds match - case Some((first, second)) => - val advanced = span0.advance(fileMapping.pieceLength, first) - val span = second.fold(advanced) { second => - advanced.take(fileMapping.pieceLength, second - first) - } - val subRangeFirst = first - val subRangeLast = second.getOrElse(file.length - 1) - Response( - status = Status.PartialContent, - headers = Headers( - Header.AcceptRanges.Bytes, - Header.ContentType(mediaType), - Header.Custom("content-range", s"bytes $subRangeFirst-$subRangeLast/${file.length}"), - ), - body = responseBody(dataStream(span), file.length - subRangeFirst), - ) - case None => - val filename = file.path.lastOption.getOrElse(s"file-$fileIndex") - Response( - status = Status.Ok, - headers = Headers( - Header.AcceptRanges.Bytes, - Header.ContentType(mediaType), - Header.Custom("content-disposition", s"""inline; filename="$filename""""), - ), - body = responseBody(dataStream(span0), file.length), - ) - } - else - ZIO.succeed( - Response.error(Status.NotFound, s"Torrent does not contain file with index $fileIndex") - ) + def responseBody(stream: Stream[IO, Byte], contentLength: Long) = + Body.fromStream(stream.translate(ioToRIO).toZStream(), contentLength) + val mediaType = + extension.flatMap(MediaType.forFileExtension).getOrElse(MediaType.application.`octet-stream`) + val span0 = fileMapping.value(fileIndex) + val rangeBounds: Option[(Long, Option[Long])] = + rangeOpt.flatMap { + case Header.Range.Single("bytes", first, second) => Some((first, second)) + case Header.Range.Prefix("bytes", first) => Some((first, None)) + case _ => None + } + rangeBounds match + case Some((first, second)) => + val advanced = span0.advance(fileMapping.pieceLength, first) + val span = second.fold(advanced)(second => advanced.take(fileMapping.pieceLength, second - first)) + val subRangeFirst = first + val subRangeLast = second.getOrElse(file.length - 1) + Response( + status = Status.PartialContent, + headers = Headers( + Header.AcceptRanges.Bytes, + Header.ContentType(mediaType), + Header.Custom("content-range", s"bytes $subRangeFirst-$subRangeLast/${file.length}"), + ), + body = responseBody(dataStream(span), file.length - subRangeFirst), + ) + case None => + val filename = file.path.lastOption.getOrElse(s"file-$fileIndex") + Response( + status = Status.Ok, + headers = Headers( + Header.AcceptRanges.Bytes, + Header.ContentType(mediaType), + Header.Custom("content-disposition", s"""inline; filename="$filename""""), + ), + body = responseBody(dataStream(span0), file.length), + ) + } case None => ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) } - + yield Routes( Method.GET / "" -> handler(Response.text("Success")), Method.GET / "ws" -> handler(handleSocket.toResponse), Method.GET / "torrent" / string("infoHash") / "metadata" -> handler { (infoHash: String, _: Request) => InfoHash.fromString.lift(infoHash) match - case Some(infoHash) => - handleGetTorrent(infoHash) - case None => - ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) + case Some(infoHash) => handleGetTorrent(infoHash) + case None => ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) }, Method.GET / "torrent" / string("infoHash") / "data" / int("fileIndex") -> handler { (infoHash: String, fileIndex: Int, req: Request) => @@ -245,23 +222,14 @@ object Main extends IOApp { ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) }, ) - } - - def serve(bindPort: Int, routes: Routes[Any, Response])(using Dispatcher[IO]): IO[Nothing] = - Server - .serve(routes) - .provide(Server.defaultWithPort(bindPort)) - .toEffect[IO] case class PieceDownloadTimeout(index: Long) extends Throwable(s"Timeout downloading piece $index") - def mkLogger: IO[Logger[IO]] = asyncScope[IO] { - import org.legogroup.woof.{*, given} - import org.legogroup.woof.slf4j.* + def mkLogger: IO[Logger[IO]] = + import org.legogroup.woof.{Filter, JsonPrinter, LogLevel, Output, DefaultLogger, Printer, given} given Filter = Filter.atLeastLevel(LogLevel.Info) given Printer = JsonPrinter() - !DefaultLogger.makeIo(Output.fromConsole[IO]) - } + DefaultLogger.makeIo(Output.fromConsole[IO]) extension (self: FileMapping.FileSpan) { def advance(pieceLength: Long, begin: Long): FileMapping.FileSpan = From aef27f94f58c5955d6bb2544b0d9ad0139d3d7af Mon Sep 17 00:00:00 2001 From: Vibe Nuage Agent Date: Sun, 11 Oct 2026 14:59:38 +0000 Subject: [PATCH 3/3] Migrate server modules to ZIO and add --no-dht smoke test flag TorrentRegistry, ServerTorrent, SocketSession, MetadataRegistry, Multiplexer, PieceStore, and TorrentIndex are now ZIO-native; cats effect remains only at the bittorrent library boundary via the new Interop helper. A new --no-dht argument skips DHT bootstrap with an empty PeerDiscovery so the HTTP and WebSocket layer can be smoke tested without network access. Co-authored-by: lavrov --- server/src/main/scala/Interop.scala | 18 ++ server/src/main/scala/Main.scala | 172 ++++++++------- server/src/main/scala/MetadataRegistry.scala | 81 +++---- server/src/main/scala/Multiplexer.scala | 85 ++++---- server/src/main/scala/PieceStore.scala | 86 ++++---- server/src/main/scala/ServerTorrent.scala | 168 +++++++-------- server/src/main/scala/SocketSession.scala | 211 +++++++------------ server/src/main/scala/TorrentIndex.scala | 40 ++-- server/src/main/scala/TorrentRegistry.scala | 199 ++++++----------- 9 files changed, 455 insertions(+), 605 deletions(-) create mode 100644 server/src/main/scala/Interop.scala diff --git a/server/src/main/scala/Interop.scala b/server/src/main/scala/Interop.scala new file mode 100644 index 0000000..7c288f2 --- /dev/null +++ b/server/src/main/scala/Interop.scala @@ -0,0 +1,18 @@ +import cats.arrow.FunctionK +import cats.effect.{IO, Resource} +import cats.effect.std.Dispatcher +import zio.interop.fromEffect +import zio.{RIO, Scope, ZIO} + +object Interop { + + def ioToRIO(using Dispatcher[IO]): FunctionK[IO, RIO[Any, _]] = + new FunctionK[IO, RIO[Any, _]] { + def apply[A](fa: IO[A]): RIO[Any, A] = fromEffect(fa) + } + + def acquireCats[A](resource: Resource[IO, A])(using Dispatcher[IO]): ZIO[Scope, Throwable, A] = + fromEffect(resource.allocated).flatMap { (value, release) => + ZIO.addFinalizer(fromEffect(release).orDie).as(value) + } +} diff --git a/server/src/main/scala/Main.scala b/server/src/main/scala/Main.scala index 3bfd23f..2e67f60 100644 --- a/server/src/main/scala/Main.scala +++ b/server/src/main/scala/Main.scala @@ -10,7 +10,6 @@ import com.github.torrentdam.bittorrent.InfoHash import com.github.torrentdam.bittorrent.dht.Node import com.github.torrentdam.bittorrent.wire.Connection import com.github.torrentdam.bittorrent.{FileMapping, PeerId, TorrentFile} -import fs2.Stream import org.legogroup.woof.{Logger, given} import org.legogroup.woof.slf4j.registerSlf4j import sun.misc.Signal @@ -18,32 +17,29 @@ import zio.http.* import zio.http.codec.PathCodec.{int, string} import zio.interop.fromEffect import zio.stream.interop.fs2z.* -import zio.{Chunk, RIO, Runtime, Scope, ZIO, ZIOAppArgs, ZIOAppDefault, durationInt} -import cps.* -import cps.syntax.* -import cps.monads.catsEffect.{*, given} +import zio.{Chunk, RIO, Scope, Task, ZIO, ZIOAppArgs, ZIOAppDefault, durationInt} +import zio.stream.ZStream +import zio.Duration as ZDuration import scala.concurrent.duration.* object Main extends ZIOAppDefault { - def downloadPieceTimeout: FiniteDuration = 3.minutes + def downloadPieceTimeout: ZDuration = 3.minutes def maxPrefetchBytes = 50 * 1000 * 1000 - given bootstrapRuntime: Runtime[Any] = Runtime.default - def run: ZIO[Environment & ZIOAppArgs & Scope, Any, Any] = for args <- getArgs.map(_.toList) + (dhtEnabled, dhtBootstrapNode) = parseArgs(args) dispatcher <- acquireDispatcher given Dispatcher[IO] = dispatcher - dhtBootstrapNode <- parseBootstrapNode(args.headOption) logger <- fromEffect(mkLogger) given Logger[IO] = logger _ <- fromEffect(logger.registerSlf4j) _ <- registerSignalHandler bindPort = Option(System.getenv("PORT")).flatMap(_.toIntOption).getOrElse(9999) - routes <- makeRoutes(dhtBootstrapNode) + routes <- makeRoutes(dhtEnabled, dhtBootstrapNode) _ <- Server.serve(routes).provide(Server.defaultWithPort(bindPort)) yield () @@ -54,6 +50,16 @@ object Main extends ZIOAppDefault { ZIO.succeed(release.unsafeRunAndForget()(using cats.effect.unsafe.IORuntime.global)) }.map(_._1) + def parseArgs(args: List[String]): (Boolean, Option[SocketAddress[Host]]) = + val argv = args.filter(_ != "--") + argv match + case "--no-dht" :: rest => + (false, rest.headOption.flatMap(SocketAddress.fromString(_))) + case first :: rest => + (true, SocketAddress.fromString(first).orElse(rest.headOption.flatMap(SocketAddress.fromString(_)))) + case Nil => + (true, None) + def parseBootstrapNode(addressString: Option[String]): ZIO[Any, Throwable, Option[SocketAddress[Host]]] = addressString match case Some(value) => @@ -64,105 +70,86 @@ object Main extends ZIOAppDefault { case None => ZIO.succeed(None) - def acquireCats[A](resource: Resource[IO, A])(using d: Dispatcher[IO] = null): ZIO[Scope, Throwable, A] = - for - allocated <- fromEffect(resource.allocated) - (value, release) = allocated - _ <- ZIO.addFinalizer(fromEffect(release).orDie) - yield value - def registerSignalHandler: ZIO[Any, Nothing, Unit] = ZIO.succeed { Signal.handle(new Signal("INT"), _ => System.exit(0)) } def resources( + dhtEnabled: Boolean, dhtBootstrapNode: Option[SocketAddress[Host]] - )(using Dispatcher[IO], Logger[IO]): Resource[IO, (TorrentRegistry, ServerTorrent.Create, TorrentIndex, MetadataRegistry[IO])] = + )(using Dispatcher[IO], Logger[IO]): ZIO[Scope, Throwable, (TorrentRegistry, ServerTorrent.Create, TorrentIndex, MetadataRegistry)] = for - random <- Resource.eval(Random.scalaUtilRandom[IO]) - selfId <- Resource.eval(PeerId.generate) - dhtNode <- Node(port = none, bootstrapNodeAddress = dhtBootstrapNode) - metadataRegistry <- Resource.eval(MetadataRegistry[IO]()) + _ <- fromEffect(Random.scalaUtilRandom[IO]) + selfId <- fromEffect(PeerId.generate) + peerDiscovery <- + if dhtEnabled then + Interop.acquireCats(Node(port = none, bootstrapNodeAddress = dhtBootstrapNode)).map(_.discovery) + else + ZIO.succeed(emptyPeerDiscovery) + metadataRegistry <- MetadataRegistry() createServerTorrent = new ServerTorrent.Create( infoHash => peerInfo => Connection.connect(selfId, peerInfo, infoHash), - dhtNode.discovery, + peerDiscovery, metadataRegistry ) - torrentRegistry <- Resource.eval(TorrentRegistry()) + torrentRegistry <- TorrentRegistry() torrentIndex <- TorrentIndex() yield (torrentRegistry, createServerTorrent, torrentIndex, metadataRegistry) - type FileIndex = Int - - def ioToRIO(using Dispatcher[IO]): cats.arrow.FunctionK[IO, RIO[Any, _]] = - new cats.arrow.FunctionK[IO, RIO[Any, _]] { - def apply[A](fa: IO[A]): RIO[Any, A] = fromEffect(fa) + def emptyPeerDiscovery: com.github.torrentdam.bittorrent.dht.PeerDiscovery = + new com.github.torrentdam.bittorrent.dht.PeerDiscovery { + def discover(infoHash: InfoHash): fs2.Stream[cats.effect.IO, com.github.torrentdam.bittorrent.PeerInfo] = + fs2.Stream.empty + def findNodes(nodeId: com.github.torrentdam.bittorrent.dht.NodeId): fs2.Stream[cats.effect.IO, com.github.torrentdam.bittorrent.dht.NodeInfo] = + fs2.Stream.empty } + type FileIndex = Int + def makeRoutes( + dhtEnabled: Boolean, dhtBootstrapNode: Option[SocketAddress[Host]] )(using Dispatcher[IO], Logger[IO]): ZIO[Scope, Throwable, Routes[Any, Response]] = for - (torrentRegistry, createServerTorrent, torrentIndex, metadataRegistry) <- acquireCats(resources(dhtBootstrapNode)) + (torrentRegistry, createServerTorrent, torrentIndex, metadataRegistry) <- resources(dhtEnabled, dhtBootstrapNode) makeTorrent: SocketSession.MakeTorrent = (infoHash, trackers) => - torrentRegistry.getOrCreate(infoHash)(createServerTorrent(infoHash, trackers)) + ZIO.scoped { + torrentRegistry.getOrCreate(infoHash)(createServerTorrent(infoHash, trackers)).flatten + } handleSocket = SocketSession(makeTorrent, metadataRegistry, torrentIndex) handleGetTorrent = (infoHash: InfoHash) => - fromEffect { - torrentRegistry - .get(infoHash) - .use { torrent => - val metadata = torrent.metadata - val torrentFile = TorrentFile(metadata, None) - val bcode = summon[BencodeFormat[TorrentFile]].write(torrentFile).toOption.get - val filename = metadata.parsed.files match - case file :: Nil if file.path.nonEmpty => file.path.last - case _ => infoHash.bytes.toHex - OptionT.liftF(IO.pure((s"$filename.torrent", encode(bcode).toByteArray))) - } - .value - }.map { - case Some((filename, bytes)) => - Response( - status = Status.Ok, - headers = Headers(Header.Custom("content-disposition", s"""inline; filename="$filename"""")), - body = Body.fromChunk(Chunk.fromArray(bytes)), - ) - case None => - Response.error(Status.NotFound, "Torrent not found") - }.catchAll(_ => ZIO.succeed(Response.status(Status.InternalServerError))) + ZIO.scoped(torrentRegistry.get(infoHash)) + .flatMap { + case Some(torrent) => + ZIO.succeed { + val metadata = torrent.metadata + val torrentFile = TorrentFile(metadata, None) + val bcode = summon[BencodeFormat[TorrentFile]].write(torrentFile).toOption.get + val filename = metadata.parsed.files match + case file :: Nil if file.path.nonEmpty => file.path.last + case _ => infoHash.bytes.toHex + Response( + status = Status.Ok, + headers = Headers(Header.Custom("content-disposition", s"""inline; filename="$filename.torrent"""")), + body = Body.fromChunk(Chunk.fromArray(encode(bcode).toByteArray)), + ) + } + case None => + ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) + } + .catchAll(_ => ZIO.succeed(Response.status(Status.InternalServerError))) handleGetData = (infoHash: InfoHash, fileIndex: FileIndex, rangeOpt: Option[Header.Range]) => - fromEffect(torrentRegistry.get(infoHash).allocated.value) - .catchAll(_ => ZIO.succeed(None: Option[(ServerTorrent, TorrentRegistry.Optional[Unit])])) - .flatMap { - case Some((torrent: ServerTorrent, release)) => + ZIO.scoped { + torrentRegistry.get(infoHash).flatMap { + case Some(torrent) => ZIO.succeed { val file = torrent.metadata.parsed.files(fileIndex) val extension = file.path.lastOption.map(_.reverse.takeWhile(_ != '.').reverse) val fileMapping = torrent.files - val parallelPieces = scala.math.max(maxPrefetchBytes / torrent.metadata.parsed.pieceLength, 2).toInt - def downloadPiece(index: Long): IO[(Long, Stream[IO, Byte])] = - torrent - .piece(index.toInt) - .timeoutTo(downloadPieceTimeout, IO.raiseError(PieceDownloadTimeout(index))) - .tupleLeft(index) - def dataStream(span: FileMapping.FileSpan): Stream[IO, Byte] = - Stream - .emits(span.beginIndex to span.endIndex) - .covary[IO] - .parEvalMap(parallelPieces)(downloadPiece) - .flatMap { - case (beginIndex, bytes) if beginIndex == span.beginIndex => bytes.drop(span.beginOffset) - case (endIndex, bytes) if endIndex == span.endIndex => bytes.take(span.endOffset) - case (_, bytes) => bytes - } - .onFinalize(release.value.void) - def responseBody(stream: Stream[IO, Byte], contentLength: Long) = - Body.fromStream(stream.translate(ioToRIO).toZStream(), contentLength) val mediaType = extension.flatMap(MediaType.forFileExtension).getOrElse(MediaType.application.`octet-stream`) val span0 = fileMapping.value(fileIndex) @@ -170,7 +157,7 @@ object Main extends ZIOAppDefault { rangeOpt.flatMap { case Header.Range.Single("bytes", first, second) => Some((first, second)) case Header.Range.Prefix("bytes", first) => Some((first, None)) - case _ => None + case _ => None } rangeBounds match case Some((first, second)) => @@ -185,7 +172,7 @@ object Main extends ZIOAppDefault { Header.ContentType(mediaType), Header.Custom("content-range", s"bytes $subRangeFirst-$subRangeLast/${file.length}"), ), - body = responseBody(dataStream(span), file.length - subRangeFirst), + body = responseBody(torrent, span, file.length - subRangeFirst), ) case None => val filename = file.path.lastOption.getOrElse(s"file-$fileIndex") @@ -196,12 +183,14 @@ object Main extends ZIOAppDefault { Header.ContentType(mediaType), Header.Custom("content-disposition", s"""inline; filename="$filename""""), ), - body = responseBody(dataStream(span0), file.length), + body = responseBody(torrent, span0, file.length), ) } case None => ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) } + } + .catchAll(_ => ZIO.succeed(Response.status(Status.InternalServerError))) yield Routes( Method.GET / "" -> handler(Response.text("Success")), @@ -210,7 +199,7 @@ object Main extends ZIOAppDefault { (infoHash: String, _: Request) => InfoHash.fromString.lift(infoHash) match case Some(infoHash) => handleGetTorrent(infoHash) - case None => ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) + case None => ZIO.succeed(Response.error(Status.NotFound, "Torrent not found")) }, Method.GET / "torrent" / string("infoHash") / "data" / int("fileIndex") -> handler { (infoHash: String, fileIndex: Int, req: Request) => @@ -223,7 +212,26 @@ object Main extends ZIOAppDefault { }, ) - case class PieceDownloadTimeout(index: Long) extends Throwable(s"Timeout downloading piece $index") + def responseBody(torrent: ServerTorrent, span: FileMapping.FileSpan, contentLength: Long)(using + Dispatcher[IO] + ): Body = + val parallelPieces = + scala.math.max(maxPrefetchBytes / torrent.metadata.parsed.pieceLength, 2).toInt + val dataStream = + ZStream + .fromIterable(span.beginIndex to span.endIndex) + .mapZIOPar(parallelPieces)(index => + torrent + .piece(index.toInt) + .flatMap(_.runFold(Chunk.empty[Byte])((acc, b) => acc :+ b)) + .map((index.toLong, _)) + ) + .flatMap { case (index, bytes: Chunk[Byte]) => + if index == span.beginIndex then ZStream.fromChunk(bytes.drop(span.beginOffset.toInt)) + else if index == span.endIndex then ZStream.fromChunk(bytes.take(span.endOffset.toInt)) + else ZStream.fromChunk(bytes) + } + Body.fromStream(dataStream, contentLength) def mkLogger: IO[Logger[IO]] = import org.legogroup.woof.{Filter, JsonPrinter, LogLevel, Output, DefaultLogger, Printer, given} diff --git a/server/src/main/scala/MetadataRegistry.scala b/server/src/main/scala/MetadataRegistry.scala index 646fd4a..0d025d6 100644 --- a/server/src/main/scala/MetadataRegistry.scala +++ b/server/src/main/scala/MetadataRegistry.scala @@ -1,34 +1,23 @@ -import cats.Monad -import cats.implicits.* -import cats.effect.implicits.* -import cats.effect.Concurrent -import cats.effect.kernel.Ref import com.github.torrentdam.bittorrent.InfoHash import com.github.torrentdam.bittorrent.TorrentMetadata.Lossless -import fs2.Stream -import fs2.concurrent.Topic - -trait MetadataRegistry[F[_]] { - - def recent: F[Iterable[(InfoHash, Lossless)]] - - def get(infoHash: InfoHash): F[Option[Lossless]] - - def put(infoHash: InfoHash, metadata: Lossless): F[Boolean] - - def subscribe: Stream[F, (InfoHash, Lossless)] +import zio.{Hub, Ref, Task, UIO, ZIO} +import zio.stream.ZStream + +trait MetadataRegistry { + def recent: Task[List[(InfoHash, Lossless)]] + def get(infoHash: InfoHash): Task[Option[Lossless]] + def put(infoHash: InfoHash, metadata: Lossless): Task[Boolean] + def subscribe: ZStream[Any, Nothing, (InfoHash, Lossless)] } object MetadataRegistry { case class State(map: Map[InfoHash, Lossless], list: List[(InfoHash, Lossless)]) { - def put(infoHash: InfoHash, metadata: Lossless): Option[State] = - if map.contains(infoHash) then none - else State(map.updated(infoHash, metadata), (infoHash, metadata) :: list).some + if map.contains(infoHash) then None + else Some(State(map.updated(infoHash, metadata), (infoHash, metadata) :: list)) def get(infoHash: InfoHash): Option[Lossless] = map.get(infoHash) - def recent: List[(InfoHash, Lossless)] = list } @@ -36,36 +25,24 @@ object MetadataRegistry { def empty: State = State(Map.empty, List.empty) } - def apply[F[_]: Concurrent](): F[MetadataRegistry[F]] = + def apply(): UIO[MetadataRegistry] = for - ref <- Ref.of[F, State](State.empty) - topic <- Topic[F, (InfoHash, Lossless)] - yield { - - new MetadataRegistry[F] { - - def recent: F[Iterable[(InfoHash, Lossless)]] = - ref.get.map(_.recent) - - def get(infoHash: InfoHash): F[Option[Lossless]] = - ref.get.map(_.get(infoHash)) - - def put(infoHash: InfoHash, metadata: Lossless): F[Boolean] = - ref - .modify { map => - map.put(infoHash, metadata) match - case Some(state) => (state, true) - case None => (map, false) - } - .flatTap { success => - Monad[F].whenA(success) { - topic.publish1((infoHash, metadata)).start - } - } - - def subscribe: Stream[F, (InfoHash, Lossless)] = topic.subscribe(1) - } - - } - + ref <- Ref.make(State.empty) + hub <- Hub.unbounded[(InfoHash, Lossless)] + yield new Impl(ref, hub) + + private class Impl(ref: Ref[State], hub: Hub[(InfoHash, Lossless)]) extends MetadataRegistry { + def recent: Task[List[(InfoHash, Lossless)]] = ref.get.map(_.recent) + def get(infoHash: InfoHash): Task[Option[Lossless]] = ref.get.map(_.get(infoHash)) + def put(infoHash: InfoHash, metadata: Lossless): Task[Boolean] = + for + success <- ref.modify { state => + state.put(infoHash, metadata) match + case Some(updated) => (true, updated) + case None => (false, state) + } + _ <- if success then hub.publish((infoHash, metadata)) else ZIO.unit + yield success + def subscribe: ZStream[Any, Nothing, (InfoHash, Lossless)] = ZStream.fromHub(hub) + } } diff --git a/server/src/main/scala/Multiplexer.scala b/server/src/main/scala/Multiplexer.scala index 8cc2b41..6f18eaf 100644 --- a/server/src/main/scala/Multiplexer.scala +++ b/server/src/main/scala/Multiplexer.scala @@ -1,54 +1,53 @@ -import cats.effect.{Async, IO} -import cats.effect.kernel.{Deferred, Ref} -import cats.implicits.* -import cats.effect.implicits.* -import fs2.Stream -import org.legogroup.woof.{Logger, given} +import zio.{Chunk, Promise, Ref, Task, UIO, ZIO} +import zio.stream.ZStream trait Multiplexer { - def get(index: Int): IO[Multiplexer.Result] + def get(index: Int): Task[ZStream[Any, Throwable, Byte]] } object Multiplexer { - type Result = Stream[IO, Byte] - - def apply(request: Int => IO[Stream[IO, Byte]])(using Logger[IO]): IO[Multiplexer] = { + def apply(request: Int => Task[Chunk[Byte]]): UIO[Multiplexer] = for - pieces <- IO.ref(Map.empty[Int, Deferred[IO, Either[Throwable, Result]]]) - yield new Multiplexer { - def get(index: Int): IO[Stream[IO, Byte]] = { - def cleanup = - pieces.update { pieces => - pieces.removed(index) - } - def requestAndCache = IO.uncancelable { _ => - for - (deferred, effect) <- pieces.modify { pieces => - pieces.get(index) match - case Some(deferred) => (pieces, (deferred, IO.unit)) - case _ => - val deferred = Deferred.unsafe[IO, Either[Throwable, Result]] - val updated = pieces.updated(index, deferred) - val effect = - request(index) - .attempt - .flatTap(deferred.complete) - .onError(e => Logger[IO].error(s"Could not download piece $index: ${e.getMessage}")) - .guarantee(cleanup) - .start - .void - (updated, (deferred, effect)) - } - _ <- effect - yield deferred - } + pieces <- Ref.make(Map.empty[Int, Promise[Throwable, Chunk[Byte]]]) + yield new Impl(pieces, request) + + private class Impl( + pieces: Ref[Map[Int, Promise[Throwable, Chunk[Byte]]]], + request: Int => Task[Chunk[Byte]] + ) extends Multiplexer { + + def get(index: Int): Task[ZStream[Any, Throwable, Byte]] = + for + deferred <- requestAndCache(index) + result <- deferred.await + yield ZStream.fromChunk(result) + + private def requestAndCache(index: Int): Task[Promise[Throwable, Chunk[Byte]]] = + ZIO.uninterruptibleMask { restore => for - deferred <- requestAndCache - result <- deferred.get - byteStream <- IO.fromEither(result) - yield byteStream + existing <- pieces.get + result <- existing.get(index) match + case Some(deferred) => ZIO.succeed(deferred) + case None => + for + deferred <- Promise.make[Throwable, Chunk[Byte]] + _ <- pieces.update(_.updated(index, deferred)) + _ <- restore( + request(index) + .tapBoth( + error => deferred.fail(error), + chunk => deferred.succeed(chunk) + ) + .ignore + .forkDaemon + ) + _ <- ensureCleanup(index) + yield deferred + yield result } - } + + private def ensureCleanup(index: Int): UIO[Unit] = + pieces.update(_.removed(index)) } } diff --git a/server/src/main/scala/PieceStore.scala b/server/src/main/scala/PieceStore.scala index f84ae9c..6bfbe18 100644 --- a/server/src/main/scala/PieceStore.scala +++ b/server/src/main/scala/PieceStore.scala @@ -1,63 +1,53 @@ -import java.nio.file.Path - -import cats.implicits.* -import cats.effect.kernel.Ref -import cats.effect.{Concurrent, Resource, Async} -import fs2.{Stream, Chunk} -import fs2.io.file.Files +import java.nio.file.{Files as JFiles, Path} import scodec.bits.ByteVector +import zio.{Chunk, Ref, Scope, Task, ZIO} -import scala.collection.immutable.BitSet -import scala.jdk.StreamConverters.* +import java.nio.file.StandardOpenOption.* -trait PieceStore[F[_]] { - def get(index: Int): F[Option[Stream[F, Byte]]] - def put(index: Int, bytes: ByteVector): F[Stream[F, Byte]] +trait PieceStore { + def get(index: Int): Task[Option[Chunk[Byte]]] + def put(index: Int, bytes: ByteVector): Task[Chunk[Byte]] } object PieceStore { - def disk[F[_]]( - directory: Path - )(using Files[F], Concurrent[F]): Resource[F, PieceStore[F]] = { - - val createDirectory = - Files[F].createDirectories(directory) - - val deleteDirectory = - Files[F].deleteDirectoryRecursively(directory) - Resource.make(createDirectory)(_ => deleteDirectory).evalMap { directory => + def disk(directory: Path): ZIO[Scope, Throwable, PieceStore] = + for + _ <- ZIO.attempt(JFiles.createDirectories(directory)) + _ <- ZIO.addFinalizer { + ZIO.attempt { + JFiles.walk(directory).sorted(java.util.Comparator.reverseOrder[Path]()).forEach(p => JFiles.deleteIfExists(p)) + }.ignore + } + availability <- Ref.make(Set.empty[Int]) + yield Impl(directory, availability) + + private case class Impl(directory: Path, availability: Ref[Set[Int]]) extends PieceStore { + def get(index: Int): Task[Option[Chunk[Byte]]] = for - availability <- Ref.of(BitSet.empty) - yield new Impl(directory, availability) - } - } - - private class Impl[F[_]](directory: Path, availability: Ref[F, BitSet])( - using - Files[F], - Concurrent[F], - ) extends PieceStore[F] { - - def get(index: Int): F[Option[Stream[F, Byte]]] = - for - availability <- availability.get - available = availability(index) - yield if available then readFile(pieceFile(index)).some else none - - def put(index: Int, bytes: ByteVector): F[Stream[F, Byte]] = { - val file = pieceFile(index) - val byteStream = Stream.chunk[F, Byte](Chunk.byteVector(bytes)) + available <- availability.get + result <- if available(index) then + ZIO + .attempt(pieceFile(index)) + .flatMap(file => + ZIO + .attempt(JFiles.readAllBytes(file)) + .map(bytes => Some(Chunk.fromArray(bytes))) + ) + else ZIO.succeed(None) + yield result + + def put(index: Int, bytes: ByteVector): Task[Chunk[Byte]] = + val chunk = Chunk.fromArray(bytes.toArray) for - _ <- Files[F].writeAll(file)(byteStream).compile.drain + _ <- ZIO.attempt { + val file = pieceFile(index) + JFiles.write(file, bytes.toArray, CREATE, TRUNCATE_EXISTING, WRITE) + } _ <- availability.update(_ + index) - yield readFile(file) - } + yield chunk private def pieceFile(index: Int) = directory.resolve(index.toString) - - private def readFile(file: Path) = - Files[F].readAll(file, 1024 * 1024) } } diff --git a/server/src/main/scala/ServerTorrent.scala b/server/src/main/scala/ServerTorrent.scala index 0d46558..36b2ce6 100644 --- a/server/src/main/scala/ServerTorrent.scala +++ b/server/src/main/scala/ServerTorrent.scala @@ -1,20 +1,20 @@ import java.nio.file.Paths -import ServerTorrent.Phase.FetchingMetadata +import cats.effect.IO +import cats.effect.std.Dispatcher import cats.syntax.all.* -import cats.effect.kernel.{Deferred, Resource} -import cats.effect.{Concurrent, IO, Resource} -import com.github.torrentdam.bittorrent.{FileMapping, InfoHash, MagnetLink, PeerInfo} +import com.github.torrentdam.bittorrent.{FileMapping, InfoHash, PeerInfo} import com.github.torrentdam.bittorrent.wire.{Connection, DownloadMetadata, Swarm, Torrent} import com.github.torrentdam.bittorrent.TorrentMetadata.Lossless -import com.github.torrentdam.bittorrent.dht.PeerDiscovery -import fs2.Stream -import fs2.concurrent.Signal import org.legogroup.woof.{Logger, given} +import zio.stream.ZStream +import zio.stream.interop.fs2z.* +import zio.interop.fromEffect +import zio.{Chunk, Promise, Scope, Task, UIO, ZIO} trait ServerTorrent { def files: FileMapping - def stats: IO[ServerTorrent.Stats] - def piece(index: Int): IO[Stream[IO, Byte]] + def stats: Task[ServerTorrent.Stats] + def piece(index: Int): Task[ZStream[Any, Throwable, Byte]] def metadata: Lossless } @@ -22,38 +22,25 @@ object ServerTorrent { sealed trait Phase object Phase { - case class PeerDiscovery(done: IO[FetchingMetadata]) extends Phase - case class FetchingMetadata(fromPeers: Signal[IO, Int], done: IO[Ready]) extends Phase + case class PeerDiscovery(done: Task[FetchingMetadata]) extends Phase + case class FetchingMetadata(fromPeers: ZStream[Any, Nothing, Int], done: Task[Ready]) extends Phase case class Ready(infoHash: InfoHash, serverTorrent: ServerTorrent) extends Phase } case class Error() extends Throwable - private def create(torrent: Torrent, pieceStore: PieceStore[IO])(using Logger[IO]): IO[ServerTorrent] = { - - def fetch(index: Int): IO[Stream[IO, Byte]] = { - for - bytes <- pieceStore.get(index) - bytes <- bytes match { - case Some(bytes) => IO.pure(bytes) - case None => - for - bytes <- torrent.downloadPiece(index) - bytes <- pieceStore.put(index, bytes) - yield bytes - } - yield bytes - } - + private def create( + torrentStats: Task[Torrent.Stats], + downloadPiece: Int => Task[Chunk[Byte]], + selfMetadata: Lossless + ): UIO[ServerTorrent] = for - multiplexer <- Multiplexer(fetch) - yield { - new ServerTorrent { - def files: FileMapping = FileMapping.fromMetadata(torrent.metadata.parsed) - def stats: IO[Stats] = - for - stats <- torrent.stats - yield Stats( + multiplexer <- Multiplexer(downloadPiece) + yield new ServerTorrent { + def files: FileMapping = FileMapping.fromMetadata(metadata.parsed) + def stats: Task[Stats] = + torrentStats.map { stats => + Stats( connected = stats.connected, availability = files.value.map { span => val range = span.beginIndex.toInt to span.endIndex.toInt @@ -61,75 +48,74 @@ object ServerTorrent { available.toDouble / range.size } ) - def piece(index: Int): IO[Stream[IO, Byte]] = multiplexer.get(index) - def metadata: Lossless = torrent.metadata - } + } + def piece(index: Int): Task[ZStream[Any, Throwable, Byte]] = multiplexer.get(index) + def metadata: Lossless = selfMetadata } - } class Create( - connect: InfoHash => PeerInfo => Resource[IO, Connection], - peerDiscovery: PeerDiscovery, - metadataRegistry: MetadataRegistry[IO] - )( - using - logger: Logger[IO], + connect: InfoHash => PeerInfo => cats.effect.Resource[IO, Connection], + peerDiscovery: com.github.torrentdam.bittorrent.dht.PeerDiscovery, + metadataRegistry: MetadataRegistry + )(using + Dispatcher[IO], + Logger[IO] ) { - - def apply(infoHash: InfoHash, trackers: List[String]): Resource[IO, Phase.PeerDiscovery] = - - def createInPhases(peerDiscoveryOutcome: FallibleDeferred[IO, Phase.FetchingMetadata]): Resource[IO, ServerTorrent] = + def apply(infoHash: InfoHash, trackers: List[String]): ZIO[Scope, Throwable, Phase.PeerDiscovery] = + for + peerDiscoveryDone <- Promise.make[Throwable, Phase.FetchingMetadata] + fiber <- createInPhases(peerDiscoveryDone, infoHash).forkDaemon + yield Phase.PeerDiscovery(peerDiscoveryDone.await) + + private def createInPhases( + peerDiscoveryOutcome: Promise[Throwable, Phase.FetchingMetadata], + infoHash: InfoHash + ): ZIO[Any, Throwable, Unit] = + ZIO.scoped { for - swarm <- Swarm( - peerDiscovery.discover(infoHash), - connect(infoHash), - ) - fetchingMetadataDone <- Resource.eval(FallibleDeferred[IO, Phase.Ready]) - _ <- Resource.eval( - peerDiscoveryOutcome.complete(FetchingMetadata(swarm.connected.count, fetchingMetadataDone.get)) + swarm <- Interop.acquireCats { + Swarm( + peerDiscovery.discover(infoHash).covary[IO], + connect(infoHash) + ) + } + fetchingMetadataDone <- Promise.make[Throwable, Phase.Ready] + countSignal: fs2.concurrent.Signal[cats.effect.IO, Int] = swarm.connected.count + fromPeersZ = countSignal.continuous.translate(Interop.ioToRIO).toZStream() + _ <- peerDiscoveryOutcome.succeed( + Phase.FetchingMetadata(fromPeersZ.catchAll(_ => ZStream.empty), fetchingMetadataDone.await) ) - metadata <- Resource.eval( - metadataRegistry.get(infoHash).flatMap { - case Some(value) => value.pure[IO] - case None => DownloadMetadata(swarm).flatTap(metadataRegistry.put(infoHash, _)) - } + metadata <- metadataRegistry.get(infoHash).flatMap { + case Some(value) => ZIO.succeed(value) + case None => fromEffect(DownloadMetadata(swarm)) + } + _ <- fromEffect(summon[Logger[IO]].info(s"Metadata downloaded")) + _ <- metadataRegistry.put(infoHash, metadata) + torrent <- Interop.acquireCats(Torrent.make(metadata, swarm)) + pieceStore <- PieceStore.disk(Paths.get("/tmp", s"bittorrent-${infoHash.toString}")) + serverTorrent <- create( + fromEffect(torrent.stats), + index => fetchPiece(torrent, pieceStore, index), + metadata ) - _ <- Resource.eval(logger.info(s"Metadata downloaded")) - torrent <- Torrent.make(metadata, swarm) - pieceStore <- PieceStore.disk[IO](Paths.get(s"/tmp", s"bittorrent-${infoHash.toString}")) - serverTorrent <- Resource.eval(create(torrent, pieceStore)) - _ <- Resource.eval(fetchingMetadataDone.complete(Phase.Ready(infoHash, serverTorrent))) - yield serverTorrent + _ <- fetchingMetadataDone.succeed(Phase.Ready(infoHash, serverTorrent)) + yield () + } + private def fetchPiece(torrent: Torrent, pieceStore: PieceStore, index: Int): Task[Chunk[Byte]] = for - peerDiscoveryDone <- Resource.eval(FallibleDeferred[IO, Phase.FetchingMetadata]) - _ <- createInPhases(peerDiscoveryDone).useForever.background - yield Phase.PeerDiscovery(peerDiscoveryDone.get) - - end apply + cached <- pieceStore.get(index) + result <- cached match + case Some(bytes) => ZIO.succeed(bytes) + case None => + fromEffect(torrent.downloadPiece(index.toLong)).flatMap { vector => + pieceStore.put(index, vector).map(_ => Chunk.fromArray(vector.toArray)) + } + yield result } case class Stats( connected: Int, availability: List[Double] ) - - trait FallibleDeferred[F[_], A] { - def complete(a: A): F[Unit] - def fail(e: Throwable): F[Unit] - def get: F[A] - } - - object FallibleDeferred { - def apply[F[_], A](using F: Concurrent[F]): F[FallibleDeferred[F, A]] = { - for - underlying <- Deferred[F, Either[Throwable, A]] - yield new FallibleDeferred[F, A] { - def complete(a: A): F[Unit] = underlying.complete(a.asRight).void - def fail(e: Throwable): F[Unit] = underlying.complete(e.asLeft).void - def get: F[A] = underlying.get.flatMap(F.fromEither) - } - } - } - } diff --git a/server/src/main/scala/SocketSession.scala b/server/src/main/scala/SocketSession.scala index 8492b0e..b1b78ff 100644 --- a/server/src/main/scala/SocketSession.scala +++ b/server/src/main/scala/SocketSession.scala @@ -1,17 +1,12 @@ -import cats.effect.{Concurrent, IO, Resource} -import cats.effect.std.Supervisor -import cats.implicits.* +import cats.effect.std.Dispatcher import com.github.torrentdam.bittorrent.InfoHash import com.github.lavrov.bittorrent.app.protocol.{Message, Event, Command} import org.legogroup.woof.{Logger, given} import zio.http.{ChannelEvent, Handler, WebSocketApp, WebSocketFrame} import zio.interop.catz.implicits.rts -import zio.interop.ToEffectSyntax -import zio.interop.fromEffect -import zio.ZIO -import cps.* -import cps.syntax.* -import cps.monads.catsEffect.{*, given} +import zio.interop.{fromEffect, ToEffectSyntax} +import zio.{Scope, Task, UIO, ZIO, durationInt} +import zio.Duration as ZDuration import scala.concurrent.duration.* import scala.util.Try @@ -20,147 +15,99 @@ object SocketSession { def apply( makeTorrent: MakeTorrent, - metadataRegistry: MetadataRegistry[IO], + metadataRegistry: MetadataRegistry, torrentIndex: TorrentIndex )(using - F: Concurrent[IO], - dispatcher: cats.effect.std.Dispatcher[IO], - logger: Logger[IO] + dispatcher: Dispatcher[cats.effect.IO], + logger: Logger[cats.effect.IO] ): WebSocketApp[Any] = Handler.webSocket { channel => - val send: Message => ZIO[Any, Nothing, Unit] = { message => + val send: Message => UIO[Unit] = { message => channel .send(ChannelEvent.Read(WebSocketFrame.Text(upickle.default.write(message)))) .catchAll(_ => ZIO.unit) } - val sendEvent: Event => IO[Unit] = { event => - send(event).toEffect[IO] + val sendEvent: Event => cats.effect.IO[Unit] = { event => + send(event).toEffect[cats.effect.IO] } - val allocatedIO: IO[(CommandHandler, IO[Unit])] = - CommandHandler(sendEvent, makeTorrent, metadataRegistry, torrentIndex).allocated - fromEffect(logger.info("Session started")) *> fromEffect(allocatedIO).flatMap { allocated => - val handler = allocated._1 - val closeHandler = allocated._2 - def process: Message => ZIO[Any, Nothing, Unit] = { - case Message.Ping => - send(Message.Pong) - case command: Command => - fromEffect { - async[IO] { - !logger.debug(s"Received $command") - !handler.handle(command) - } - }.catchAll(_ => ZIO.unit) - case _ => - ZIO.unit - } - val receiveLoop = channel.receiveAll { - case ChannelEvent.Read(WebSocketFrame.Text(text)) => - JsonMessage.lift(text).map(process).getOrElse(ZIO.unit) - case _ => - ZIO.unit - } - val pingLoop = ZIO.sleep(zio.Duration.fromScala(10.seconds)) *> - channel.send(ChannelEvent.Read(WebSocketFrame.Ping)).catchAll(_ => ZIO.unit) - receiveLoop - .zipPar(pingLoop.forever) - .ensuring { - fromEffect { - closeHandler *> logger.info("Session closed") - }.catchAll(_ => ZIO.unit) + val process: Message => UIO[Unit] = { + case Message.Ping => + send(Message.Pong) + case command: Command => + fromEffect { + summon[Logger[cats.effect.IO]].debug(s"Received $command") } + .catchAll(_ => ZIO.unit) *> + handleCommand(command, makeTorrent, sendEvent) + case _ => + ZIO.unit } + val receiveLoop = channel.receiveAll { + case ChannelEvent.Read(WebSocketFrame.Text(text)) => + JsonMessage.lift(text).map(process).getOrElse(ZIO.unit) + case _ => + ZIO.unit + } + val pingLoop = ZIO.sleep(ZDuration.fromScala(10.seconds)) *> + channel.send(ChannelEvent.Read(WebSocketFrame.Ping)).catchAll(_ => ZIO.unit) + receiveLoop + .zipPar(pingLoop.forever) + .ensuring(ZIO.unit) } - private val JsonMessage: PartialFunction[String, Message] = - ((input: String) => Try(upickle.default.read[Message](input)).toOption).unlift - - class CommandHandler( - send: Event => IO[Unit], + private def handleCommand( + command: Command, makeTorrent: MakeTorrent, - metadataRegistry: MetadataRegistry[IO], - torrentIndex: TorrentIndex, - supervisor: Supervisor[IO] - )(using - F: Concurrent[IO], - logger: Logger[IO] - ) { - - def handle(command: Command): IO[Unit] = - command match - case Message.RequestTorrent(infoHash, trackers) => - for - _ <- send(Message.RequestAccepted(infoHash)) - _ <- handleRequestTorrent(InfoHash(infoHash.bytes), trackers) - yield () - - private def handleRequestTorrent(infoHash: InfoHash, trackers: List[String]): IO[Unit] = - supervisor.supervise( - makeTorrent(infoHash, trackers) - .use { getTorrent => - getTorrent - .flatMap { (phase: ServerTorrent.Phase.PeerDiscovery) => - phase.done - } - .flatMap { (phase: ServerTorrent.Phase.FetchingMetadata) => - phase.fromPeers.discrete - .evalTap { count => - send(Message.TorrentPeersDiscovered(infoHash, count)) - } - .interruptWhen(phase.done.void.attempt) - .compile - .drain >> - phase.done - } - .flatMap { (phase: ServerTorrent.Phase.Ready) => - val metadata = phase.serverTorrent.metadata.parsed - val files = metadata.files.map(f => Message.File(f.path, f.length)) - send(Message.TorrentMetadataReceived(infoHash, metadata.name, files)) >> - phase.serverTorrent.pure[IO] - } - .timeout(5.minutes) - .flatMap { torrent => - sendTorrentStats(infoHash, torrent) - } - } - .orElse { - send(Message.TorrentError(infoHash, "Could not fetch metadata")) - } - ).void + send: Event => cats.effect.IO[Unit] + )(using dispatcher: Dispatcher[cats.effect.IO]): UIO[Unit] = + command match + case Message.RequestTorrent(infoHash, trackers) => + val infoHash0 = InfoHash(infoHash.bytes) + fromEffect(send(Message.RequestAccepted(infoHash))) + .catchAll(_ => ZIO.unit) *> + handleRequestTorrent(infoHash0, trackers, makeTorrent, send).forkDaemon.unit - private def sendTorrentStats(infoHash: InfoHash, torrent: ServerTorrent): IO[Nothing] = - val sendStats = - for - stats <- torrent.stats - _ <- send(Message.TorrentStats(infoHash, stats.connected, stats.availability)) - _ <- IO.sleep(5.seconds) - yield () - sendStats.foreverM - } + private def handleRequestTorrent( + infoHash: InfoHash, + trackers: List[String], + makeTorrent: MakeTorrent, + send: Event => cats.effect.IO[Unit] + )(using dispatcher: Dispatcher[cats.effect.IO]): UIO[Unit] = + val outcome = + for + peerDiscovery <- makeTorrent(infoHash, trackers) + fetchingMetadata <- peerDiscovery.done + fromPeersStream = fetchingMetadata.fromPeers + _ <- fromPeersStream + .mapZIO(count => + fromEffect(send(Message.TorrentPeersDiscovered(infoHash, count))).catchAll(_ => ZIO.unit) + ) + .runDrain + .forkDaemon + ready <- fetchingMetadata.done + _ <- sendStatsLoop(infoHash, ready, send) + yield () + outcome.catchAll { _ => + fromEffect(send(Message.TorrentError(infoHash, "Could not fetch metadata"))).catchAll(_ => ZIO.unit) + }.ignore - object CommandHandler { - def apply( - send: Event => IO[Unit], - makeTorrent: MakeTorrent, - metadataRegistry: MetadataRegistry[IO], - torrentIndex: TorrentIndex - )(using - F: Concurrent[IO], - logger: Logger[IO] - ): Resource[IO, CommandHandler] = + private def sendStatsLoop( + infoHash: InfoHash, + phase: ServerTorrent.Phase.Ready, + send: Event => cats.effect.IO[Unit] + )(using dispatcher: Dispatcher[cats.effect.IO]): UIO[Unit] = + val loop = for - supervisor <- Supervisor[IO] - yield - new CommandHandler( - send, - makeTorrent, - metadataRegistry, - torrentIndex, - supervisor - ) - } + stats <- phase.serverTorrent.stats + _ <- fromEffect(send(Message.TorrentStats(infoHash, stats.connected, stats.availability))) + _ <- ZIO.sleep(ZDuration.fromScala(5.seconds)) + yield () + loop.forever.catchAll(_ => ZIO.unit) + + private val JsonMessage: PartialFunction[String, Message] = + ((input: String) => Try(upickle.default.read[Message](input)).toOption).unlift trait MakeTorrent { - def apply(infoHash: InfoHash, trackers: List[String]): Resource[IO, IO[ServerTorrent.Phase.PeerDiscovery]] + def apply(infoHash: InfoHash, trackers: List[String]): Task[ServerTorrent.Phase.PeerDiscovery] } } diff --git a/server/src/main/scala/TorrentIndex.scala b/server/src/main/scala/TorrentIndex.scala index 2d5a9a3..ea06212 100644 --- a/server/src/main/scala/TorrentIndex.scala +++ b/server/src/main/scala/TorrentIndex.scala @@ -1,49 +1,37 @@ -import cats.effect.{IO, Resource} -import cats.effect.kernel.Ref -import cats.syntax.all.* -import org.legogroup.woof.{Logger, given} - -import scala.concurrent.duration.* -import scala.util.chaining.scalaUtilChainingOps +import zio.{Ref, Task, UIO, ZIO} trait TorrentIndex { import TorrentIndex.Entry - - def search(text: String): IO[List[Entry]] + def search(text: String): Task[List[Entry]] } object TorrentIndex { - def apply()(using logger: Logger[IO]): Resource[IO, TorrentIndex] = { - Resource.eval { - for - ref <- Ref.of[IO, Index](Index()) - yield - impl(ref.get) - } - } + def apply(): UIO[TorrentIndex] = + for + ref <- Ref.make(Index()) + yield impl(ref.get) - private def impl(entries: IO[Index]): TorrentIndex = { + private def impl(entries: Task[Index]): TorrentIndex = new TorrentIndex { - def search(text: String): IO[List[Entry]] = { + def search(text: String): Task[List[Entry]] = { val words = text.toLowerCase.split(' ') - for index <- entries + for + index <- entries yield index.entries .view - .map { - case (searchField, entry) => - val score = words.map(word => if searchField.contains(word) then word.length else 0).sum - (entry, score) + .map { case (searchField, entry) => + val score = words.map(word => if searchField.contains(word) then word.length else 0).sum + (entry, score) } .filter(_._2 > 0) .toList - .sortBy(_._2)(using Ordering[Int].reverse) + .sortBy(-_._2) .take(100) .map(_._1) } } - } case class Index(entries: List[(String, Entry)] = List.empty) case class Entry(name: String, infoHash: String, size: Long, ext: List[String]) diff --git a/server/src/main/scala/TorrentRegistry.scala b/server/src/main/scala/TorrentRegistry.scala index 17c9dea..a1ed72b 100644 --- a/server/src/main/scala/TorrentRegistry.scala +++ b/server/src/main/scala/TorrentRegistry.scala @@ -1,168 +1,105 @@ -import cats.Show.Shown - -import java.nio.file.Paths -import cats.data.OptionT -import cats.effect.* -import cats.effect.kernel.{Deferred, Ref} -import cats.implicits.* import com.github.torrentdam.bittorrent.InfoHash import org.legogroup.woof.{Logger, given} +import zio.{Promise, Ref, Scope, Task, UIO, ZIO} -import scala.concurrent.duration.* +import java.util.concurrent.TimeUnit trait TorrentRegistry { - import TorrentRegistry.Optional - def getOrCreate(infoHash: InfoHash)(createTorrent: Resource[IO, ServerTorrent.Phase.PeerDiscovery]): Resource[IO, IO[ServerTorrent.Phase.PeerDiscovery]] - def get(infoHash: InfoHash): Resource[Optional, ServerTorrent] + def getOrCreate(infoHash: InfoHash)(createTorrent: ZIO[Scope, Throwable, ServerTorrent.Phase.PeerDiscovery]) + : ZIO[Scope, Throwable, Task[ServerTorrent.Phase.PeerDiscovery]] + def get(infoHash: InfoHash): ZIO[Scope, Throwable, Option[ServerTorrent]] } object TorrentRegistry { - type Optional[A] = OptionT[IO, A] - - def apply()(using logger: Logger[IO]): IO[TorrentRegistry] = - for - ref <- Ref.of[IO, Registry](emptyRegistry) - yield new Impl(ref) - case class UsageCountingCell( - get: IO[ServerTorrent.Phase.PeerDiscovery], + get: Task[ServerTorrent.Phase.PeerDiscovery], ready: Option[ServerTorrent], - close: IO[Unit], + close: UIO[Unit], count: Int, usedCount: Int ) - private type Registry = Map[InfoHash, UsageCountingCell] - private val emptyRegistry: Registry = Map.empty + def apply(): UIO[TorrentRegistry] = + for + ref <- Ref.make(Map.empty[InfoHash, UsageCountingCell]) + yield Impl(ref) - private class Impl(ref: Ref[IO, Registry])(using - logger: Logger[IO] - ) extends TorrentRegistry { - import Logger.withLogContext + private class Impl(ref: Ref[Map[InfoHash, UsageCountingCell]]) extends TorrentRegistry { + import zio.interop.catz.implicits.rts def getOrCreate( infoHash: InfoHash - )( - createTorrent: Resource[IO, ServerTorrent.Phase.PeerDiscovery] - ): Resource[IO, IO[ServerTorrent.Phase.PeerDiscovery]] = - Resource { - ref - .modify { registry => - registry.get(infoHash) match { - case Some(cell) => - val updatedCell = cell.copy(count = cell.count + 1, usedCount = cell.usedCount + 1) - val updatedRegistry = registry.updated(infoHash, updatedCell) - (updatedRegistry, Right(updatedCell.get)) - case None => - val torrentDeferred = Deferred.unsafe[IO, Either[Throwable, ServerTorrent.Phase.PeerDiscovery]] - val closeDeferred = Deferred.unsafe[IO, Unit] - val getTorrent = torrentDeferred.get.flatMap(IO.fromEither) - val closeTorrent = closeDeferred.complete(()).void - val createdCell = UsageCountingCell(getTorrent, none, closeTorrent, 1, 1) - val updatedRegistry = registry.updated(infoHash, createdCell) - (updatedRegistry, Left((createdCell.get, torrentDeferred.complete, closeDeferred.get))) - } - } - .flatMap { - case Right(get) => - logger.debug(s"Found existing torrent") >> - IO.pure((get, release(infoHash))) - case Left((get, complete, waitClosed)) => - logger.info(s"Make new torrent") >> - createTorrent - .use(phase => - complete(phase.asRight) >> - cacheReadyPhase(phase).start >> - waitClosed - ) - .handleErrorWith { e => - logger.error(s"Could not create torrent $e") >> - complete(e.asLeft) - } - .start - .as((get, release(infoHash))) - } - .withLogContext("infoHash", infoHash.toString) - } + )(createTorrent: ZIO[Scope, Throwable, ServerTorrent.Phase.PeerDiscovery]) + : ZIO[Scope, Throwable, Task[ServerTorrent.Phase.PeerDiscovery]] = + for + existing <- ref.modify { registry => + registry.get(infoHash) match + case Some(cell) => + val updatedCell = cell.copy(count = cell.count + 1, usedCount = cell.usedCount + 1) + (Some(updatedCell.get), registry.updated(infoHash, updatedCell)) + case None => (None, registry) + } + get <- existing match + case Some(get) => ZIO.succeed(get) + case None => + for + torrentDeferred <- Promise.make[Throwable, ServerTorrent.Phase.PeerDiscovery] + closeDeferred <- Promise.make[Nothing, Unit] + get = torrentDeferred.await + closeTorrent = closeDeferred.succeed(()).unit + cell = UsageCountingCell(get, None, closeTorrent, 1, 1) + _ <- ref.update(_.updated(infoHash, cell)) + fiber <- createTorrent + .flatMap(phase => torrentDeferred.succeed(phase) *> cacheReadyPhase(phase)) + .catchAll(e => torrentDeferred.fail(e)) + .forkDaemon + yield get + yield get - def get(infoHash: InfoHash): Resource[Optional, ServerTorrent] = - Resource { - for - torrent <- OptionT( - ref - .modify { registry => - registry.get(infoHash) match { - case Some(cell) if cell.ready.isDefined => - val updatedCell = cell.copy(count = cell.count + 1, usedCount = cell.usedCount + 1) - val updatedRegistry = registry.updated(infoHash, updatedCell) - (updatedRegistry, cell.ready) - case _ => - (registry, none) - } - } - ) - _ <- logger.debug(s"Found existing torrent").to[Optional] - yield (torrent, release(infoHash).to[Optional]) - } + def get(infoHash: InfoHash): ZIO[Scope, Throwable, Option[ServerTorrent]] = + for + torrent <- ref.modify { registry => + registry.get(infoHash) match + case Some(cell) if cell.ready.isDefined => + val updatedCell = cell.copy(count = cell.count + 1, usedCount = cell.usedCount + 1) + (cell.ready, registry.updated(infoHash, updatedCell)) + case _ => (None, registry) + } + yield torrent - private def cacheReadyPhase(peerDiscovery: ServerTorrent.Phase.PeerDiscovery): IO[Unit] = { + private def cacheReadyPhase(peerDiscovery: ServerTorrent.Phase.PeerDiscovery): Task[Unit] = for fetchingMetadata <- peerDiscovery.done ready <- fetchingMetadata.done _ <- ref.update { registry => - registry.updatedWith(ready.infoHash)( - _.map( - _.copy(ready = ready.serverTorrent.some) - ) - ) + registry.updatedWith(ready.infoHash)(_.map(_.copy(ready = Some(ready.serverTorrent)))) } yield () - } - private def release(infoHash: InfoHash)(using logger: Logger[IO]): IO[Unit] = - logger.debug(s"Release torrent") >> - ref - .modify { registry => + private def release(infoHash: InfoHash): UIO[Unit] = + for + cell <- ref.modify { registry => val cell = registry(infoHash) val updatedCell = cell.copy(count = cell.count - 1) - (registry.updated(infoHash, updatedCell), updatedCell) - } - .flatMap { cell => - if cell.count == 0 then - scheduleClose(infoHash, _.usedCount == cell.usedCount) - else - logger.debug(s"Torrent is still in use ${cell.count}") + (updatedCell, registry.updated(infoHash, updatedCell)) } + _ <- if cell.count == 0 then scheduleClose(infoHash, _.usedCount == cell.usedCount) + else ZIO.unit + yield () - private def scheduleClose(infoHash: InfoHash, closeIf: UsageCountingCell => Boolean)(using - logger: Logger[IO] - ): IO[Unit] = { - val idleTimeout = 30.minutes - val waitAndTry = - logger.debug(s"Schedule torrent closure in $idleTimeout") >> - IO.sleep(idleTimeout) >> - tryClose(infoHash, closeIf) - waitAndTry.start.void - } + private def scheduleClose(infoHash: InfoHash, closeIf: UsageCountingCell => Boolean): UIO[Unit] = + val idleTimeout = zio.Duration(30, TimeUnit.MINUTES) + (for + _ <- ZIO.sleep(idleTimeout) + _ <- tryClose(infoHash, closeIf) + yield ()).forkDaemon.unit - private def tryClose(infoHash: InfoHash, closeIf: UsageCountingCell => Boolean)(using - logger: Logger[IO] - ): IO[Unit] = { + private def tryClose(infoHash: InfoHash, closeIf: UsageCountingCell => Boolean): UIO[Unit] = ref.modify { registry => - registry.get(infoHash) match { - case Some(cell) if closeIf(cell) => - ( - registry.removed(infoHash), - cell.close >> logger.info(s"Closed torrent") - ) - case _ => - ( - registry, - IO.unit - ) - } + registry.get(infoHash) match + case Some(cell) if closeIf(cell) => (cell.close, registry.removed(infoHash)) + case _ => (ZIO.unit, registry) }.flatten - } } }