<commit_before>package sttp.client3.asynchttpclient

import java.nio.ByteBuffer

import io.netty.handler.codec.http.HttpHeaders
import org.asynchttpclient.AsyncHandler.State
import org.asynchttpclient.handler.StreamedAsyncHandler
import org.asynchttpclient.proxy.ProxyServer
import org.asynchttpclient.ws.{WebSocketListener, WebSocketUpgradeHandler, WebSocket => AHCWebSocket}
import org.asynchttpclient.{
  AsyncHandler,
  AsyncHttpClient,
  BoundRequestBuilder,
  DefaultAsyncHttpClient,
  DefaultAsyncHttpClientConfig,
  HttpResponseBodyPart,
  HttpResponseStatus,
  Realm,
  RequestBuilder,
  Request => AsyncRequest,
  Response => AsyncResponse
}
import org.reactivestreams.{Publisher, Subscriber, Subscription}
import sttp.capabilities.{Effect, Streams}
import sttp.client3
import sttp.client3.SttpBackendOptions.ProxyType.{Http, Socks}
import sttp.client3.internal.ws.{SimpleQueue, WebSocketEvent}
import sttp.monad.syntax._
import sttp.monad.{Canceler, MonadAsyncError, MonadError}
import sttp.client3.{Response, SttpBackend, SttpBackendOptions, _}
import sttp.model._

import scala.collection.JavaConverters._
import scala.collection.immutable.Seq
import scala.util.Try

abstract class AsyncHttpClientBackend[F[_], S <: Streams[S], P](
    asyncHttpClient: AsyncHttpClient,
    private implicit val monad: MonadAsyncError[F],
    closeClient: Boolean,
    customizeRequest: BoundRequestBuilder => BoundRequestBuilder
) extends SttpBackend[F, P] {

  val streams: Streams[S]
  type PE = P with Effect[F]

  override def send[T, R >: PE](r: Request[T, R]): F[Response[T]] =
    adjustExceptions(r) {
      preparedRequest(r).flatMap { ahcRequest =>
        if (r.isWebSocket) sendWebSocket(r, ahcRequest) else sendRegular(r, ahcRequest)
      }
    }

  private def sendRegular[T, R >: PE](r: Request[T, R], ahcRequest: BoundRequestBuilder): F[Response[T]] = {
    monad.flatten(monad.async[F[Response[T]]] { cb =>
      def success(r: F[Response[T]]): Unit = cb(Right(r))
      def error(t: Throwable): Unit = cb(Left(t))

      val lf = ahcRequest.execute(streamingAsyncHandler(r, success, error))
      Canceler(() => lf.abort(new InterruptedException))
    })
  }

  private def sendWebSocket[T, R >: PE](
      r: Request[T, R],
      ahcRequest: BoundRequestBuilder
  ): F[Response[T]] =
    createSimpleQueue[WebSocketEvent].flatMap { queue =>
      monad.flatten(monad.async[F[Response[T]]] { cb =>
        val initListener =
          new WebSocketInitListener(r, queue, (r: F[Response[T]]) => cb(Right(r)), t => cb(Left(t)))
        val h = new WebSocketUpgradeHandler.Builder()
          .addWebSocketListener(initListener)
          .build()

        val lf = ahcRequest.execute(h)

        Canceler(() => lf.abort(new InterruptedException))
      })
    }

  override def responseMonad: MonadError[F] = monad

  protected def bodyFromAHC: BodyFromAHC[F, S]
  protected def bodyToAHC: BodyToAHC[F, S]<commit_msg>Use Future.cancel instead of ListenableFuture.abort in async-http-client<commit_after>package sttp.client3.asynchttpclient

import java.nio.ByteBuffer

import io.netty.handler.codec.http.HttpHeaders
import org.asynchttpclient.AsyncHandler.State
import org.asynchttpclient.handler.StreamedAsyncHandler
import org.asynchttpclient.proxy.ProxyServer
import org.asynchttpclient.ws.{WebSocketListener, WebSocketUpgradeHandler, WebSocket => AHCWebSocket}
import org.asynchttpclient.{
  AsyncHandler,
  AsyncHttpClient,
  BoundRequestBuilder,
  DefaultAsyncHttpClient,
  DefaultAsyncHttpClientConfig,
  HttpResponseBodyPart,
  HttpResponseStatus,
  Realm,
  RequestBuilder,
  Request => AsyncRequest,
  Response => AsyncResponse
}
import org.reactivestreams.{Publisher, Subscriber, Subscription}
import sttp.capabilities.{Effect, Streams}
import sttp.client3
import sttp.client3.SttpBackendOptions.ProxyType.{Http, Socks}
import sttp.client3.internal.ws.{SimpleQueue, WebSocketEvent}
import sttp.monad.syntax._
import sttp.monad.{Canceler, MonadAsyncError, MonadError}
import sttp.client3.{Response, SttpBackend, SttpBackendOptions, _}
import sttp.model._

import scala.collection.JavaConverters._
import scala.collection.immutable.Seq
import scala.util.Try

abstract class AsyncHttpClientBackend[F[_], S <: Streams[S], P](
    asyncHttpClient: AsyncHttpClient,
    private implicit val monad: MonadAsyncError[F],
    closeClient: Boolean,
    customizeRequest: BoundRequestBuilder => BoundRequestBuilder
) extends SttpBackend[F, P] {

  val streams: Streams[S]
  type PE = P with Effect[F]

  override def send[T, R >: PE](r: Request[T, R]): F[Response[T]] =
    adjustExceptions(r) {
      preparedRequest(r).flatMap { ahcRequest =>
        if (r.isWebSocket) sendWebSocket(r, ahcRequest) else sendRegular(r, ahcRequest)
      }
    }

  private def sendRegular[T, R >: PE](r: Request[T, R], ahcRequest: BoundRequestBuilder): F[Response[T]] = {
    monad.flatten(monad.async[F[Response[T]]] { cb =>
      def success(r: F[Response[T]]): Unit = cb(Right(r))
      def error(t: Throwable): Unit = cb(Left(t))

      val lf = ahcRequest.execute(streamingAsyncHandler(r, success, error))
      Canceler(() => lf.cancel(true))
    })
  }

  private def sendWebSocket[T, R >: PE](
      r: Request[T, R],
      ahcRequest: BoundRequestBuilder
  ): F[Response[T]] =
    createSimpleQueue[WebSocketEvent].flatMap { queue =>
      monad.flatten(monad.async[F[Response[T]]] { cb =>
        val initListener =
          new WebSocketInitListener(r, queue, (r: F[Response[T]]) => cb(Right(r)), t => cb(Left(t)))
        val h = new WebSocketUpgradeHandler.Builder()
          .addWebSocketListener(initListener)
          .build()

        val lf = ahcRequest.execute(h)

        Canceler(() => lf.cancel(true))
      })
    }

  override def responseMonad: MonadError[F] = monad

  protected def bodyFromAHC: BodyFromAHC[F, S]
  protected def bodyToAHC: BodyToAHC[F, S]
{'fraction_non_alphanumeric': 0.11762706102328743, 'fraction_numerical': 0.0016998130205677375, 'mean_word_length': 4.514526710402999, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 33, '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': '1571296', 'n_tokens_mistral': 1958, 'n_tokens_neox': 1932, 'n_words': 391}