<commit_before>package sttp.client.asynchttpclient.zio

import sttp.client.asynchttpclient.WebSocketHandler
import sttp.client.asynchttpclient.internal.NativeWebSocketHandler
import sttp.client.impl.zio.TaskMonadAsyncError
import sttp.client.ws.internal.AsyncQueue
import sttp.client.ws.{WebSocket, WebSocketEvent}
import sttp.model.ws.WebSocketBufferFull
import zio.{Queue, Runtime, Task, UIO, ZIO}

object ZioWebSocketHandler {
  private class ZioAsyncQueue[A](queue: Queue[A], runtime: Runtime[Any]) extends AsyncQueue[Task, A] {
    override def offer(t: A): Unit = {
      if (!runtime.unsafeRun(queue.offer(t))) {
        throw new WebSocketBufferFull()
      }
    }
    override def poll: Task[A] = {
      queue.take
    }
  }

  /**
    * Creates a new [[WebSocketHandler]] which should be used *once* to send and receive from a single websocket.
    *
    * The handler will internally buffer incoming messages, and expose an instance of the [[WebSocket]] interface for
    * sending/receiving messages.
    *
    * @param incomingBufferCapacity Should the buffer of incoming websocket events be bounded. If yes, unreceived
    *                               events will some point cause the websocket to error and close. If no, unreceived
    *                               messages will eventually take up all available memory.
    */
  def apply(incomingBufferCapacity: Option[Int] = None): UIO[WebSocketHandler[WebSocket[Task]]] = {
    val queue = incomingBufferCapacity match {
      case Some(capacity) => Queue.dropping[WebSocketEvent](capacity)<commit_msg>Use a def instead of a val<commit_after>package sttp.client.asynchttpclient.zio

import sttp.client.asynchttpclient.WebSocketHandler
import sttp.client.asynchttpclient.internal.NativeWebSocketHandler
import sttp.client.impl.zio.TaskMonadAsyncError
import sttp.client.ws.internal.AsyncQueue
import sttp.client.ws.{WebSocket, WebSocketEvent}
import sttp.model.ws.WebSocketBufferFull
import zio.{Queue, Runtime, Task, UIO, ZIO}

object ZioWebSocketHandler {
  private class ZioAsyncQueue[A](queue: Queue[A], runtime: Runtime[Any]) extends AsyncQueue[Task, A] {
    override def offer(t: A): Unit = {
      if (!runtime.unsafeRun(queue.offer(t))) {
        throw new WebSocketBufferFull()
      }
    }
    override def poll: Task[A] = {
      queue.take
    }
  }

  /**
    * Returns an effect, which creates a new [[WebSocketHandler]]. The handler should be used *once* to send and
    * receive from a single websocket.
    *
    * The handler will internally buffer incoming messages, and expose an instance of the [[WebSocket]] interface for
    * sending/receiving messages.
    *
    * @param incomingBufferCapacity Should the buffer of incoming websocket events be bounded. If yes, unreceived
    *                               events will some point cause the websocket to error and close. If no, unreceived
    *                               messages will eventually take up all available memory.
    */
  def apply(incomingBufferCapacity: Option[Int] = None): UIO[WebSocketHandler[WebSocket[Task]]] = {
    def queue = incomingBufferCapacity match {
      case Some(capacity) => Queue.dropping[WebSocketEvent](capacity)
{'fraction_non_alphanumeric': 0.08824446523230434, 'fraction_numerical': 0.0, 'mean_word_length': 3.443213296398892, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4447056', 'n_tokens_mistral': 907, 'n_tokens_neox': 895, 'n_words': 270}