Simple http pipeline with aggregation

.scalafmt.conf

@@ -0,0 +1,27 @@
+version="2.7.5"
+align = none
+align.openParenCallSite = false
+align.openParenDefnSite = false
+align.tokens = []
+assumeStandardLibraryStripMargin = false
+binPack.parentConstructors = false
+continuationIndent.callSite = 2
+continuationIndent.defnSite = 2
+danglingParentheses = true
+docstrings = ScalaDoc
+docstrings.blankFirstLine = yes
+encoding = UTF-8
+importSelectors = singleLine
+includeCurlyBraceInSelectChains = true
+indentOperator = spray
+lineEndings = unix
+maxColumn = 80
+newlines.alwaysBeforeTopLevelStatements = true
+newlines.sometimesBeforeColonInMethodReturnType = false
+optIn.breakChainOnFirstMethodDot = true
+rewrite.rules = [
+  PreferCurlyFors
+]
+spaces {
+  inImportCurlyBraces = false
+}

benchmarks/src/main/scala/fs2/netty/benchmarks/echo/Fs2Netty.scala

@@ -29,7 +29,7 @@ object Fs2Netty extends IOApp {
     val port = Port(args(1).toInt).get
 
     val rsrc = Network[IO] flatMap { net =>
-      val handlers = net.server(host, port) map { client =>
+      val handlers = net.server(host, port, options = Nil) map { client =>
         client.reads.through(client.writes).attempt.void
       }
 

build.sbt

@@ -25,7 +25,7 @@ ThisBuild / organizationName := "Typelevel"
 
 ThisBuild / startYear := Some(2021)
 
-ThisBuild / crossScalaVersions := Seq("2.12.12", "2.13.4", "3.0.0-M3")
+ThisBuild / crossScalaVersions := Seq("2.12.12", "2.13.4")  // "3.0.0-M3" temporarily removed to easier/speedier
 
 ThisBuild / githubWorkflowOSes ++= Seq("macos-latest", "windows-latest")
 

core/src/main/scala/fs2/netty/Socket.scala → core/src/main/scala/fs2/netty/NettyChannelInitializer.scala

@@ -14,18 +14,20 @@
  * limitations under the License.
  */
 
-package fs2
-package netty
+package fs2.netty
 
-import com.comcast.ip4s.{IpAddress, SocketAddress}
+import fs2.netty.pipeline.socket.Socket
+import io.netty.channel.socket.SocketChannel
+import io.netty.channel.{Channel, ChannelInitializer}
 
-trait Socket[F[_]] {
+trait NettyChannelInitializer[F[_], O, I] {
 
-  def localAddress: F[SocketAddress[IpAddress]]
-  def remoteAddress: F[SocketAddress[IpAddress]]
+  def toSocketChannelInitializer(
+    cb: Socket[F, O, I] => F[Unit]
+  ): F[ChannelInitializer[SocketChannel]] =
+    toChannelInitializer[SocketChannel](cb)
 
-  def reads: Stream[F, Byte]
-
-  def write(bytes: Chunk[Byte]): F[Unit]
-  def writes: Pipe[F, Byte, INothing]
+  def toChannelInitializer[C <: Channel](
+    cb: Socket[F, O, I] => F[Unit]
+  ): F[ChannelInitializer[C]]
 }