<commit_before>package sttp.client.httpclient.monix.internal

import java.net.http.WebSocket.Listener
import java.util.concurrent.atomic.AtomicBoolean
import java.net.http.{WebSocket => JWebSocket}
import java.nio.ByteBuffer
import java.util.concurrent.{CompletableFuture, CompletionStage}
import java.util.function.BiConsumer

import sttp.client.httpclient.WebSocketHandler
import sttp.client.monad.syntax._
import sttp.client.monad.{MonadAsyncError, MonadError}
import sttp.client.ws.internal.AsyncQueue
import sttp.client.ws.{WebSocket, WebSocketEvent}
import sttp.model.ws.{WebSocketClosed, WebSocketFrame}

object NativeWebSocketHandler {
  def apply[F[_]](queue: AsyncQueue[F, WebSocketEvent], monad: MonadAsyncError[F]): WebSocketHandler[WebSocket[F]] = {
    val isOpen: AtomicBoolean = new AtomicBoolean(false)
    WebSocketHandler(
      new AddToQueueListener(queue, isOpen),
      httpClientWebSocketToWebSocket(_, queue, isOpen, monad)
    )
  }

  private def httpClientWebSocketToWebSocket[F[_]](
      ws: JWebSocket,
      queue: AsyncQueue[F, WebSocketEvent],
      _isOpen: AtomicBoolean,
      _monad: MonadAsyncError[F]
  ): WebSocket[F] = new WebSocket[F] {
    override def receive: F[Either[WebSocketEvent.Close, WebSocketFrame.Incoming]] = {
      queue.poll.flatMap {
        case WebSocketEvent.Open() => receive
        case c: WebSocketEvent.Close =>
          queue.offer(WebSocketEvent.Error(new WebSocketClosed))
          monad.unit(Left(c))
        case e @ WebSocketEvent.Error(t: Exception) =>
          // putting back the error so that subsequent invocations end in an error as well, instead of hanging
          queue.offer(e)
          monad.error(t)
        case WebSocketEvent.Error(t) => throw t
        case WebSocketEvent.Frame(f) =>
          monad.unit(Right(f)).map { frame =>
            ws.request(1)
            frame
          }
      }
    }

    override def send(f: WebSocketFrame, isContinuation: Boolean = false): F[Unit] =
      monad.flatten(monad.eval(fromCompletableFuture(f match {
        case WebSocketFrame.Text(payload, finalFragment, _) if !isContinuation =>
          ws.sendText(payload, finalFragment)
        case WebSocketFrame.Text(payload, finalFragment, _) if isContinuation =>
          ws.sendText(payload, finalFragment)
        case WebSocketFrame.Binary(payload, finalFragment, _) if !isContinuation =>
          ws.sendBinary(ByteBuffer.wrap(payload), finalFragment)
        case WebSocketFrame.Binary(payload, finalFragment, _) if isContinuation =>
          ws.sendBinary(ByteBuffer.wrap(payload), finalFragment)
        case WebSocketFrame.Ping(payload)                 => ws.sendPing(ByteBuffer.wrap(payload))
        case WebSocketFrame.Pong(payload)                 => ws.sendPong(ByteBuffer.wrap(payload))
        case WebSocketFrame.Close(statusCode, reasonText) => ws.sendClose(statusCode, reasonText)
      })))

    override def isOpen: F[Boolean] = monad.eval(_isOpen.get())
<commit_msg>Remove distinction between continuation message and normal message<commit_after>package sttp.client.httpclient.monix.internal

import java.net.http.WebSocket.Listener
import java.util.concurrent.atomic.AtomicBoolean
import java.net.http.{WebSocket => JWebSocket}
import java.nio.ByteBuffer
import java.util.concurrent.{CompletableFuture, CompletionStage}
import java.util.function.BiConsumer

import sttp.client.httpclient.WebSocketHandler
import sttp.client.monad.syntax._
import sttp.client.monad.{MonadAsyncError, MonadError}
import sttp.client.ws.internal.AsyncQueue
import sttp.client.ws.{WebSocket, WebSocketEvent}
import sttp.model.ws.{WebSocketClosed, WebSocketFrame}

object NativeWebSocketHandler {
  def apply[F[_]](queue: AsyncQueue[F, WebSocketEvent], monad: MonadAsyncError[F]): WebSocketHandler[WebSocket[F]] = {
    val isOpen: AtomicBoolean = new AtomicBoolean(false)
    WebSocketHandler(
      new AddToQueueListener(queue, isOpen),
      httpClientWebSocketToWebSocket(_, queue, isOpen, monad)
    )
  }

  private def httpClientWebSocketToWebSocket[F[_]](
      ws: JWebSocket,
      queue: AsyncQueue[F, WebSocketEvent],
      _isOpen: AtomicBoolean,
      _monad: MonadAsyncError[F]
  ): WebSocket[F] = new WebSocket[F] {
    override def receive: F[Either[WebSocketEvent.Close, WebSocketFrame.Incoming]] = {
      queue.poll.flatMap {
        case WebSocketEvent.Open() => receive
        case c: WebSocketEvent.Close =>
          queue.offer(WebSocketEvent.Error(new WebSocketClosed))
          monad.unit(Left(c))
        case e @ WebSocketEvent.Error(t: Exception) =>
          // putting back the error so that subsequent invocations end in an error as well, instead of hanging
          queue.offer(e)
          monad.error(t)
        case WebSocketEvent.Error(t) => throw t
        case WebSocketEvent.Frame(f) =>
          monad.unit(Right(f)).map { frame =>
            ws.request(1)
            frame
          }
      }
    }

    override def send(f: WebSocketFrame, isContinuation: Boolean = false): F[Unit] =
      monad.flatten(monad.eval(fromCompletableFuture(f match {
        case WebSocketFrame.Text(payload, finalFragment, _) =>
          ws.sendText(payload, finalFragment)
        case WebSocketFrame.Binary(payload, finalFragment, _) =>
          ws.sendBinary(ByteBuffer.wrap(payload), finalFragment)
        case WebSocketFrame.Ping(payload)                 => ws.sendPing(ByteBuffer.wrap(payload))
        case WebSocketFrame.Pong(payload)                 => ws.sendPong(ByteBuffer.wrap(payload))
        case WebSocketFrame.Close(statusCode, reasonText) => ws.sendClose(statusCode, reasonText)
      })))

    override def isOpen: F[Boolean] = monad.eval(_isOpen.get())

{'fraction_non_alphanumeric': 0.10508771929824562, 'fraction_numerical': 0.0003508771929824561, 'mean_word_length': 3.5138558986539983, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1571792', 'n_tokens_mistral': 1722, 'n_tokens_neox': 1676, 'n_words': 329}