Merge pull request #3257 from armanbilge/fix/js-tls-server-session

Fix TLS session access on JS
typelevel · Jul 21, 2023 · 230c96a · 230c96a
2 parents febad15 + d15242a
commit 230c96a
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 31 deletions.
diff --git a/io/js/src/main/scala/fs2/io/internal/facade/tls.scala b/io/js/src/main/scala/fs2/io/internal/facade/tls.scala
@@ -188,6 +188,8 @@ package tls {
 
     def ssl: SSL = js.native
 
+    def getSession(): js.UndefOr[Uint8Array] = js.native
+
   }
 
   @js.native

diff --git a/io/js/src/main/scala/fs2/io/net/tls/TLSContextPlatform.scala b/io/js/src/main/scala/fs2/io/net/tls/TLSContextPlatform.scala
@@ -89,8 +89,7 @@ private[tls] trait TLSContextCompanionPlatform { self: TLSContext.type =>
                             )
                         )
                         tlsSock
-                      },
-                      seqDispatcher
+                      }
                     )
                     .evalTap(_ => handshake.get.rethrow)
                 }
@@ -129,8 +128,7 @@ private[tls] trait TLSContextCompanionPlatform { self: TLSContext.type =>
                             )
                         )
                         tlsSock
-                      },
-                      seqDispatcher
+                      }
                     )
                     .evalTap(_ => verifyError.get.rethrow)
                 }

diff --git a/io/js/src/main/scala/fs2/io/net/tls/TLSSocketPlatform.scala b/io/js/src/main/scala/fs2/io/net/tls/TLSSocketPlatform.scala
@@ -26,52 +26,35 @@ package tls
 
 import cats.effect.kernel.Async
 import cats.effect.kernel.Resource
-import cats.effect.std.Dispatcher
 import cats.effect.syntax.all._
 import cats.syntax.all._
-import fs2.concurrent.SignallingRef
 import fs2.io.internal.facade
 import fs2.io.internal.SuspendedStream
 import scodec.bits.ByteVector
 
-import scala.scalajs.js
-
 private[tls] trait TLSSocketPlatform[F[_]]
 
 private[tls] trait TLSSocketCompanionPlatform { self: TLSSocket.type =>
 
   private[tls] def forAsync[F[_]](
       socket: Socket[F],
-      upgrade: fs2.io.Duplex => facade.tls.TLSSocket,
-      dispatcher: Dispatcher[F]
+      upgrade: fs2.io.Duplex => facade.tls.TLSSocket
   )(implicit F: Async[F]): Resource[F, TLSSocket[F]] =
     for {
       duplexOut <- mkDuplex(socket.reads)
       (duplex, out) = duplexOut
       _ <- out.through(socket.writes).compile.drain.background
-      sessionRef <- SignallingRef[F].of(Option.empty[SSLSession]).toResource
       tlsSockReadable <- suspendReadableAndRead(
         destroyIfNotEnded = false,
         destroyIfCanceled = false
-      ) {
-        val tlsSock = upgrade(duplex)
-        tlsSock.on[js.typedarray.Uint8Array](
-          "session",
-          session =>
-            dispatcher.unsafeRunAndForget(
-              sessionRef.set(Some(new SSLSession(ByteVector.view(session))))
-            )
-        )
-        tlsSock
-      }
+      )(upgrade(duplex))
       (tlsSock, readable) = tlsSockReadable
-      _ <- Resource.unit[F].onFinalize(F.delay(tlsSock.removeAllListeners("session")))
       readStream <- SuspendedStream(readable)
     } yield new AsyncTLSSocket(
       tlsSock,
       readStream,
       socket,
-      sessionRef.discrete.unNone.head.compile.lastOrError,
+      F.delay(new SSLSession(ByteVector.view(tlsSock.getSession().get))),
       F.delay[Any](tlsSock.alpnProtocol).flatMap {
         case false            => "".pure // mimicking JVM
         case protocol: String => protocol.pure

diff --git a/io/js/src/test/scala/fs2/io/net/tls/TLSSocketSuite.scala b/io/js/src/test/scala/fs2/io/net/tls/TLSSocketSuite.scala
@@ -269,23 +269,23 @@ class TLSSocketSuite extends TLSSuite {
           .flatMap { case (server, clientSocket) =>
             val echoServer = server
               .evalTap(s => s.applicationProtocol.assertEquals("h2"))
-              .map { socket =>
-                socket.reads.chunks.foreach(socket.write(_)) ++ Stream.exec(socket.session.void)
+              .flatMap { socket =>
+                Stream
+                  .eval(socket.session)
+                  .concurrently(socket.reads.chunks.foreach(socket.write(_)))
               }
-              .parJoinUnbounded
 
             val client = Stream.resource(clientSocket).flatMap { clientSocket =>
               Stream.exec(clientSocket.applicationProtocol.assertEquals("h2")) ++
                 Stream.exec(clientSocket.session.void) ++
                 Stream.exec(clientSocket.write(msg)) ++
-                clientSocket.reads.take(msg.size.toLong)
+                Stream.eval(clientSocket.readN(msg.size).assertEquals(msg))
             }
 
-            client.concurrently(echoServer)
+            client.parZip(echoServer)
           }
           .compile
-          .to(Chunk)
-          .assertEquals(msg)
+          .drain
       }
     }