Expose 'prefetchN' parameter in 'ServerOptions'. Move 'f.g.{client => shared}.StreamIngest'.

Author: seigert

build.sbt

@@ -41,6 +41,7 @@ inThisBuild(
       ProblemFilters.exclude[DirectMissingMethodProblem]("fs2.grpc.client.StreamIngest.create"),
       ProblemFilters.exclude[DirectMissingMethodProblem]("fs2.grpc.server.Fs2StreamServerCallListener*"),
       ProblemFilters.exclude[DirectMissingMethodProblem]("fs2.grpc.client.Fs2StreamClientCallListener*"),
+      ProblemFilters.exclude[MissingClassProblem]("fs2.grpc.client.StreamIngest*"),
       ProblemFilters.exclude[MissingClassProblem]("fs2.grpc.codegen.Fs2GrpcServicePrinter$constants$"),
       ProblemFilters.exclude[MissingFieldProblem]("fs2.grpc.codegen.Fs2GrpcServicePrinter.constants"),
       // deleted private classes

runtime/src/main/scala/fs2/grpc/client/Fs2StreamClientCallListener.scala

@@ -27,6 +27,7 @@ import cats.effect.SyncIO
 import cats.implicits._
 import cats.effect.kernel.Concurrent
 import cats.effect.std.Dispatcher
+import fs2.grpc.shared.StreamIngest
 import io.grpc.{ClientCall, Metadata, Status}
 
 private[client] class Fs2StreamClientCallListener[F[_], Response] private (

runtime/src/main/scala/fs2/grpc/server/Fs2StreamServerCallListener.scala

@@ -28,7 +28,7 @@ import cats.syntax.all._
 import cats.effect.kernel.Deferred
 import cats.effect.{Async, SyncIO}
 import cats.effect.std.Dispatcher
-import fs2.grpc.client.StreamIngest
+import fs2.grpc.shared.StreamIngest
 import io.grpc.ServerCall
 
 private[server] class Fs2StreamServerCallListener[F[_], Request, Response] private (
@@ -67,7 +67,8 @@ private[server] object Fs2StreamServerCallListener {
     )(implicit F: Async[F]): F[Fs2StreamServerCallListener[F, Request, Response]] = for {
       isCancelled <- Deferred[F, Unit]
       request = (n: Int) => F.delay(call.request(n))
-      ingest <- StreamIngest[F, Request](request, prefetchN = 1)
+      prefetchN = math.max(options.prefetchN, 1)
+      ingest <- StreamIngest[F, Request](request, prefetchN)
       serverCall <- Fs2ServerCall[F, Request, Response](call, options)
     } yield new Fs2StreamServerCallListener[F, Request, Response](
       ingest,

runtime/src/main/scala/fs2/grpc/server/ServerOptions.scala

@@ -24,12 +24,22 @@ package grpc
 package server
 
 sealed abstract class ServerOptions private (
+    val prefetchN: Int,
     val callOptionsFn: ServerCallOptions => ServerCallOptions
 ) {
 
   private def copy(
-      callOptionsFn: ServerCallOptions => ServerCallOptions
-  ): ServerOptions = new ServerOptions(callOptionsFn) {}
+      prefetchN: Int = this.prefetchN,
+      callOptionsFn: ServerCallOptions => ServerCallOptions = this.callOptionsFn
+  ): ServerOptions = new ServerOptions(prefetchN, callOptionsFn) {}
+
+  /** Prefetch up to @param n messages from a client. The server will try to keep the internal buffer filled according
+    * to the provided value.
+    *
+    * If the provided value is less than 1 it defaults to 1.
+    */
+  def withPrefetchN(n: Int): ServerOptions =
+    copy(prefetchN = math.max(n, 1))
 
   /** Function that is applied on `fs2.grpc.ServerCallOptions.default` for each new RPC call.
     */
@@ -40,6 +50,7 @@ sealed abstract class ServerOptions private (
 object ServerOptions {
 
   val default: ServerOptions = new ServerOptions(
+    prefetchN = 1,
     callOptionsFn = identity
   ) {}
 

…scala/fs2/grpc/client/StreamIngest.scala …scala/fs2/grpc/shared/StreamIngest.scalaruntime/src/main/scala/fs2/grpc/client/StreamIngest.scala renamed to runtime/src/main/scala/fs2/grpc/shared/StreamIngest.scala runtime/src/main/scala/fs2/grpc/client/StreamIngest.scala renamed to runtime/src/main/scala/fs2/grpc/shared/StreamIngest.scala

@@ -21,7 +21,7 @@
 
 package fs2
 package grpc
-package client
+package shared
 
 import cats.implicits._
 import cats.effect.{Concurrent, Ref}

…/fs2/grpc/client/StreamIngestSuite.scala …/fs2/grpc/shared/StreamIngestSuite.scalaruntime/src/test/scala/fs2/grpc/client/StreamIngestSuite.scala renamed to runtime/src/test/scala/fs2/grpc/shared/StreamIngestSuite.scala runtime/src/test/scala/fs2/grpc/client/StreamIngestSuite.scala renamed to runtime/src/test/scala/fs2/grpc/shared/StreamIngestSuite.scala

@@ -21,7 +21,7 @@
 
 package fs2
 package grpc
-package client
+package shared
 
 import cats.effect._
 import munit._