Fix Socket resource handling

Author: armanbilge

io/jvm-native/src/main/scala/fs2/io/net/SocketGroupPlatform.scala

@@ -75,7 +75,7 @@ private[net] trait SocketGroupCompanionPlatform { self: SocketGroup.type =>
           }
         }
 
-      setup.flatMap(ch => Resource.eval(connect(ch))).flatMap(Socket.forAsync(_))
+      setup.evalMap(ch => connect(ch) *> Socket.forAsync(ch))
     }
 
     def serverResource(
@@ -108,32 +108,34 @@ private[net] trait SocketGroupCompanionPlatform { self: SocketGroup.type =>
           sch: AsynchronousServerSocketChannel
       ): Stream[F, Socket[F]] = {
         def go: Stream[F, Socket[F]] = {
-          def acceptChannel: F[AsynchronousSocketChannel] =
-            Async[F].async[AsynchronousSocketChannel] { cb =>
-              Async[F]
-                .delay {
-                  sch.accept(
-                    null,
-                    new CompletionHandler[AsynchronousSocketChannel, Void] {
-                      def completed(ch: AsynchronousSocketChannel, attachment: Void): Unit =
-                        cb(Right(ch))
-                      def failed(rsn: Throwable, attachment: Void): Unit =
-                        cb(Left(rsn))
-                    }
-                  )
-                }
-                .as(Some(Async[F].delay(sch.close())))
+          def acceptChannel = Resource.makeFull[F, AsynchronousSocketChannel] { poll =>
+            poll {
+              Async[F].async[AsynchronousSocketChannel] { cb =>
+                Async[F]
+                  .delay {
+                    sch.accept(
+                      null,
+                      new CompletionHandler[AsynchronousSocketChannel, Void] {
+                        def completed(ch: AsynchronousSocketChannel, attachment: Void): Unit =
+                          cb(Right(ch))
+                        def failed(rsn: Throwable, attachment: Void): Unit =
+                          cb(Left(rsn))
+                      }
+                    )
+                  }
+                  .as(Some(Async[F].delay(sch.close())))
+              }
             }
+          }(ch => Async[F].delay(if (ch.isOpen) ch.close else ()))
 
           def setOpts(ch: AsynchronousSocketChannel) =
             Async[F].delay {
               options.foreach(o => ch.setOption(o.key, o.value))
             }
 
-          Stream.eval(acceptChannel.attempt).flatMap {
-            case Left(_) => Stream.empty[F]
-            case Right(accepted) =>
-              Stream.resource(Socket.forAsync(accepted).evalTap(_ => setOpts(accepted)))
+          Stream.resource(acceptChannel.attempt).flatMap {
+            case Left(_)         => Stream.empty[F]
+            case Right(accepted) => Stream.eval(setOpts(accepted) *> Socket.forAsync(accepted))
           } ++ go
         }
 

io/jvm-native/src/main/scala/fs2/io/net/SocketPlatform.scala

@@ -24,7 +24,7 @@ package io
 package net
 
 import com.comcast.ip4s.{IpAddress, SocketAddress}
-import cats.effect.{Async, Resource}
+import cats.effect.Async
 import cats.effect.std.Semaphore
 import cats.syntax.all._
 
@@ -35,12 +35,10 @@ import java.nio.{Buffer, ByteBuffer}
 private[net] trait SocketCompanionPlatform {
   private[net] def forAsync[F[_]: Async](
       ch: AsynchronousSocketChannel
-  ): Resource[F, Socket[F]] =
-    Resource.make {
-      (Semaphore[F](1), Semaphore[F](1)).mapN { (readSemaphore, writeSemaphore) =>
-        new AsyncSocket[F](ch, readSemaphore, writeSemaphore)
-      }
-    }(_ => Async[F].delay(if (ch.isOpen) ch.close else ()))
+  ): F[Socket[F]] =
+    (Semaphore[F](1), Semaphore[F](1)).mapN { (readSemaphore, writeSemaphore) =>
+      new AsyncSocket[F](ch, readSemaphore, writeSemaphore)
+    }
 
   private[net] abstract class BufferedReads[F[_]](
       readSemaphore: Semaphore[F]