Ember wsclient #7196

danghieutrung · 2023-07-01T13:44:28Z

Ember Web Socket Client #7261

This is my current draft PR. I will later create the WSConnection for the WSClient

yilinwei

I left some comments. I'm super excited to have this in ember, so thank you very much for looking into it!

yilinwei · 2023-07-02T16:16:16Z

ember-client/shared/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala

+            unixSockets
+              .orElse(defaultUnixSockets)
+              .liftTo(
+                new RuntimeException(


Maybe IllegalStateException or UnsupportedOperationException? It'd be nice if we didn't need to throw at all though.

I'm guessing that this is thrown if the JDK doesn't include the unix socket support or we don't have fs2 netty or something? Maybe we can have a reference to how to fix that in the exception.

Ah, I think this code is copy-pasted from elsewhere.

In any case, currently this exception is thrown only on Scala Native, and that will be solved very soon.

yilinwei · 2023-07-02T16:30:51Z

ember-core/shared/src/main/scala/org/http4s/ember/core/WebSocketHelpers.scala

+                // call it repeatedly until all frames in the buffer are decoded.
+                val frames = ArrayBuffer.empty[WebSocketFrame]
+                var byteBuffer = ByteBuffer.wrap(buffer)
+                val transcoder = new FrameTranscoder(isClient)


Maybe we can use FrameTranscoder.getMsgLength to peek at the frame rather than this try catch approach?

It will mean that we don't have to throw exceptions as part of the normal control flow, which is a little expensive.

yilinwei · 2023-07-02T18:11:44Z

ember-core/shared/src/main/scala/org/http4s/ember/core/WebSocketHelpers.scala

+          case Some((chunk, next)) =>
+            ApplicativeThrow[Pull[F, WebSocketFrame, *]]
+              .catchNonFatal { // `bufferToFrame` might throw
+                val buffer = acc ++ chunk.toArray[Byte]


I wonder if it's worth breaking up the flow into two separate flows, since I think that the single frame case is possibly common enough to optimize; I'd need to check the runtime profile of some of the websocket apps that we're using though.

diesalbla

A few suggestions

diesalbla · 2023-07-03T08:45:43Z

ember-client/shared/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala

+      tlsContextOptWithDefault <-
+        tlsContextOpt
+          .fold(Network[F].tlsContext.systemResource.attempt.map(_.toOption))(
+            _.some.pure[Resource[F, *]]
+          )


IMHO once an option fold takes more than two lines, a match may come simpler:

Suggested change

tlsContextOptWithDefault <-

tlsContextOpt

.fold(Network[F].tlsContext.systemResource.attempt.map(_.toOption))(

_.some.pure[Resource[F, *]]

)

tlsContextOptWithDefault <- tlsContextOpt match {

case None => Network[F].tlsContext.systemResource.attempt.map(_.toOption)

case Some(tlsContext) => Resource.pure(Some(tlsContext))

}

diesalbla · 2023-07-03T08:53:17Z

ember-core/shared/src/main/scala/org/http4s/ember/core/WebSocketHelpers.scala

+    Stream.eval(read).flatMap {
+      case Some(bytes) =>
+        Stream.chunk(bytes) ++ readStream(read)
+      case None => Stream.empty
+  }


I thought there was a Stream factory method for this kind of loop, F[Option[Chunk[A]]] => Stream[F, A], but there is not.

Steam.eval(read).repeat.unNoneTerminate.unchunks?

diesalbla · 2023-07-03T08:55:22Z

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/WebSocketHelpers.scala

+          def receive: F[Option[org.http4s.client.websocket.WSFrame]] = ???
+          def send(wsf: org.http4s.client.websocket.WSFrame): F[Unit] = ???
+          def sendMany[G[_], A <: org.http4s.client.websocket.WSFrame](wsfs: G[A])(implicit


Perhaps add an import org.http4s.client.websocket.WSFrame at top?

diesalbla · 2023-07-03T08:57:58Z

...ver/shared/src/test/scala/org/http4s/ember/server/internal/ExampleWebSocketClientSuite.scala

+    for {
+      _ <- emberClient
+        .use { client =>
+          client
+            .run(wsRequest(s"ws://${server.address.getHostName}:${server.address.getPort}/ws-echo"))


Suggested change

for {

_ <- emberClient

.use { client =>

client

.run(wsRequest(s"ws://${server.address.getHostName}:${server.address.getPort}/ws-echo"))

val req = s"ws://${server.address.getHostName}:${server.address.getPort}/ws-echo"

for {

_ <- emberClient

.use { client =>

client

.run(wsRequest(req))

diesalbla · 2023-07-03T09:03:05Z

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/WebSocketHelpers.scala

+        Resource.eval(validateServerHandshake(res, exampleSecWebSocketKey)).map { isValid =>
+          isValid match {
+            case Left(_) => None


If the body of a map, or flatMap, or other higher-order-function, has the form of a pattern-matching, it is possible to omit the variable and just write the cases.

Suggested change

Resource.eval(validateServerHandshake(res, exampleSecWebSocketKey)).map { isValid =>

isValid match {

case Left(_) => None

Resource.eval(validateServerHandshake(res, exampleSecWebSocketKey)).map {

case Left(_) => None

diesalbla · 2023-07-03T09:05:32Z

ember-client/shared/src/main/scala/org/http4s/ember/client/EmberClientBuilder.scala

+          ) { case ((response, drain), exitCase) =>
+            exitCase match {
+              case Resource.ExitCase.Succeeded =>


Likewise, if we have a pattern-match as the body of a case, we can flatten the pattern-matches by nesting the cases:

Suggested change

) { case ((response, drain), exitCase) =>

exitCase match {

case Resource.ExitCase.Succeeded =>

) {

case ((response, drain), Resource.ExitCase.Succeeded) =>

Thanks for all the suggestions!

armanbilge

awesome!

armanbilge · 2023-07-04T13:09:48Z

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/WebSocketHelpers.scala

-          def send(wsf: org.http4s.client.websocket.WSFrame): F[Unit] = ???
-          def sendMany[G[_], A <: org.http4s.client.websocket.WSFrame](wsfs: G[A])(implicit
+          def receive: F[Option[WSFrame]] = for {
+            _ <- socket.reads.through(decodeFrames(true)).evalTap(clientReceiveQueue.offer(_)).compile.drain


You should move this one into the for and run it with .background.

armanbilge · 2023-07-04T13:10:52Z

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/WebSocketHelpers.scala

+          } yield frame
+          def send(wsf: WSFrame): F[Unit] = for {
+            _ <- toWebSocketFrame(wsf).map(clientSendQueue.offer(_))
+            _ <- clientSendQueue.take.map(f => frameToBytes(f, true).traverse_(c => socket.write(c)))


Same with this one, I think you might need to do foreverM.background so that it loops forever.

kiendang · 2023-07-05T03:48:34Z

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/WebSocketHelpers.scala

+  def getSocket[F[_]](client: Client[F], request: Request[F])(implicit
+      F: MonadCancel[F, Throwable]
+  ): Resource[F, Resource[F, Option[Socket[F]]]] = {
+    val webSocketKey = createWebSocketKey[F]
+    client
+      .run(request)
+      .map { res =>
+        Resource
+          .eval(validateServerHandshake(res, exampleSecWebSocketKey))
+          .map(isValid => isValid.toOption *> res.attributes.lookup(webSocketKey))
+      }
+  }


The type could be simplified a bit here

Suggested change

def getSocket[F[_]](client: Client[F], request: Request[F])(implicit

F: MonadCancel[F, Throwable]

): Resource[F, Resource[F, Option[Socket[F]]]] = {

val webSocketKey = createWebSocketKey[F]

client

.run(request)

.map { res =>

Resource

.eval(validateServerHandshake(res, exampleSecWebSocketKey))

.map(isValid => isValid.toOption *> res.attributes.lookup(webSocketKey))

}

}

def getSocket[F[_]](client: Client[F], request: Request[F])(implicit

F: MonadCancel[F, Throwable]

): Resource[F, Option[Socket[F]]] = {

val webSocketKey = createWebSocketKey[F]

client

.run(request)

.flatMap { res =>

Resource

.eval(validateServerHandshake(res, exampleSecWebSocketKey))

.map(isValid => isValid.toOption *> res.attributes.lookup(webSocketKey))

}

}

kiendang · 2023-07-05T03:55:54Z

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/WebSocketHelpers.scala

+          socketResource <- getSocket(emberClient, httpWSRequest)
+          socketOption <- socketResource


Suggested change

socketResource <- getSocket(emberClient, httpWSRequest)

socketOption <- socketResource

socketOption <- getSocket(emberClient, httpWSRequest)

diesalbla · 2023-07-10T14:51:49Z

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/WebSocketHelpers.scala

+      .run(request)
+      .flatMap { res =>
+        Resource
+          .eval(validateServerHandshake(res, exampleSecWebSocketKey))
+          .map(isValid => isValid.toOption *> res.attributes.lookup(webSocketKey))
+      }


The inner Resource.eval(ioa).map(a2b) should be equivalent to a Resource.eval(ioa.map(a2b))

The outer Resource.flatMap(res => Resource.eval(resToB)) should be equivalent to a Resource.evalMap.

Also, that isValid.toOption *> is easier as an if-then-else or an ifF.

Thanks will edit this part of code!

armanbilge · 2023-07-24T19:12:46Z

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/EmberWSClient.scala

+      val httpWSRequest = Request[F]()
+        .withUri(wsRequest.uri)
+        .withHeaders(wsRequest.headers)
+        .withMethod(Method.GET)


I think there's a bug here: the Uri for wsRequest will be either ws:// or wss:// but this is not correct for a HTTP request. So we should rewrite that to http:// or https:// before making the request.

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/EmberWSClient.scala

…ernal/EmberWSClient.scala Co-authored-by: Diego E. Alonso <diesalbla@gmail.com>

armanbilge · 2023-08-18T14:53:27Z

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/EmberWSClient.scala

+        _ <- clientSendQueue.take
+          .flatMap(f => frameToBytes(f, true).traverse_(c => socket.write(c)))
+          .foreverM
+          .void
+          .background


Let's change this to an FS2 Channel so that it can have graceful closure. Then, we should close the Channel after sending a CloseFrame.

Finally, to make sure that all the frames are sent before closing the connection, we should wait for the process to complete naturally.

So something like this:

sendingFinished <- clientSendChannel .foreach(f => frameToBytes(f, true).traverse_(c => socket.write(c))) .compile .drain .background _ <- Resource.onFinalize(sendingFinished.void) // wait for sending to finish

armanbilge · 2023-08-25T18:14:16Z

Published as 0.23.23-90-3d739a9-SNAPSHOT. Please give it a try!

https://github.com/http4s/http4s/actions/runs/5977900248/attempts/2#summary-16220706487

danghieutrung added 4 commits July 1, 2023 16:02

Add build method for WebSocket

de8fbba

Add createWebSocketKey method

d1db42b

Move shared WebSocket helpers functins

09fb168

Create example WebSocket client

413049d

mergify bot added module:ember-core module:ember-server module:ember-client labels Jul 1, 2023

yilinwei reviewed Jul 2, 2023

View reviewed changes

Update getSocket function

3d51d45

diesalbla reviewed Jul 3, 2023

View reviewed changes

Update WSConnection interface

e8502f7

armanbilge reviewed Jul 4, 2023

View reviewed changes

kiendang reviewed Jul 5, 2023

View reviewed changes

Update WSConnection interface

e1a7536

diesalbla reviewed Jul 10, 2023

View reviewed changes

danghieutrung and others added 14 commits July 12, 2023 17:01

Use shared function from ember-core WebSocketHelpers

44c7add

Add headers

dd920a1

Update formatting

9c7306c

Add WSClient for Ember

223e449

Fix formatting

cec189a

Delete file

97b11cb

Add WebSocketKey object

a9b6c41

Use WebSocketKey object for buildWebSocket method

ca62628

Fix formatting

5034505

Fix formatting

95b4dff

Fix formatting

323ad4d

Testing WSConnection interface

84dd6e6

debugging. it works!

7ee439c

Fix send method

d5f84d1

armanbilge reviewed Jul 24, 2023

View reviewed changes

danghieutrung added 6 commits July 28, 2023 21:31

Update WSClient interface

7098848

Unable to receive binary frame

81f3584

Add a test for sending and receiving a binary frame

e599d10

Fix formatting

d5974d6

Add handling for Close Frame

3e7f226

Add test for receving Close Frame

4663af0

diesalbla reviewed Aug 9, 2023

View reviewed changes

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/EmberWSClient.scala Outdated Show resolved Hide resolved

diesalbla reviewed Aug 9, 2023

View reviewed changes

ember-client/shared/src/main/scala/org/http4s/ember/client/internal/EmberWSClient.scala Outdated Show resolved Hide resolved

danghieutrung and others added 7 commits August 10, 2023 09:43

Update ember-client/shared/src/main/scala/org/http4s/ember/client/int…

f0fd00b

…ernal/EmberWSClient.scala Co-authored-by: Diego E. Alonso <diesalbla@gmail.com>

Update ember-client/shared/src/main/scala/org/http4s/ember/client/int…

e8ecfb9

…ernal/EmberWSClient.scala Co-authored-by: Diego E. Alonso <diesalbla@gmail.com>

Fix formatting

a64b724

Workaround FS2 bug

0fb6ad0

Fix handling for closeFrame

5c04627

Add test for receiving closeFrame in high-level connection

c088a3e

Modify auto disconnection test

5c12d7a

armanbilge reviewed Aug 18, 2023

View reviewed changes

danghieutrung added 5 commits August 22, 2023 08:44

Replace queue with channel

d468835

Hanging while trying to receive

83aad30

Automatically send the close frame

3770c53

Add a test for automatically sending the close frame

96bde01

Fix formatting

a79466f

armanbilge mentioned this pull request Aug 22, 2023

Add Channel#sendClose typelevel/fs2#3279

Closed

armanbilge and others added 4 commits August 25, 2023 14:32

Merge branch 'ember-wsclient'

d65f0f2

Add buildWebSocket

adb63a0

Add import

f5d7bf9

Fix formatting

658e0e8

armanbilge merged commit 3d739a9 into http4s:ember-wsclient Aug 25, 2023

yurique mentioned this pull request Nov 25, 2023

WIP: Support Scala Native joan38/kubernetes-client#239

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Ember wsclient #7196

Ember wsclient #7196

danghieutrung commented Jul 1, 2023 •

edited by armanbilge

Loading

yilinwei left a comment

yilinwei Jul 2, 2023

armanbilge Jul 2, 2023

yilinwei Jul 2, 2023

yilinwei Jul 2, 2023

diesalbla left a comment

diesalbla Jul 3, 2023

diesalbla Jul 3, 2023

yilinwei Jul 21, 2023 •

edited

Loading

diesalbla Jul 3, 2023

diesalbla Jul 3, 2023

diesalbla Jul 3, 2023

diesalbla Jul 3, 2023

danghieutrung Jul 3, 2023

armanbilge left a comment

armanbilge Jul 4, 2023

armanbilge Jul 4, 2023

kiendang Jul 5, 2023

kiendang Jul 5, 2023

diesalbla Jul 10, 2023 •

edited

Loading

danghieutrung Jul 12, 2023

armanbilge Jul 24, 2023

armanbilge Aug 18, 2023 •

edited

Loading

armanbilge commented Aug 25, 2023

		socketResource <- getSocket(emberClient, httpWSRequest)
		socketOption <- socketResource

	socketResource <- getSocket(emberClient, httpWSRequest)
	socketOption <- socketResource
	socketOption <- getSocket(emberClient, httpWSRequest)

Ember wsclient #7196

Ember wsclient #7196

Conversation

danghieutrung commented Jul 1, 2023 • edited by armanbilge Loading

yilinwei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

diesalbla left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yilinwei Jul 21, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

armanbilge left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

diesalbla Jul 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

armanbilge Aug 18, 2023 • edited Loading

Choose a reason for hiding this comment

armanbilge commented Aug 25, 2023

danghieutrung commented Jul 1, 2023 •

edited by armanbilge

Loading

yilinwei Jul 21, 2023 •

edited

Loading

diesalbla Jul 10, 2023 •

edited

Loading

armanbilge Aug 18, 2023 •

edited

Loading