<commit_before>/*
 * Copyright 2014 http4s.org
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.http4s
package client
package blaze

import cats.effect._
import cats.effect.implicits._
import cats.syntax.all._
import fs2._
import java.nio.ByteBuffer
import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicReference
import org.http4s.{headers => H}
import org.http4s.Uri.{Authority, RegName}
import org.http4s.blaze.pipeline.Command.EOF
import org.http4s.blazecore.Http1Stage
import org.http4s.blazecore.util.Http1Writer
import org.http4s.headers.{Connection, Host, `Content-Length`, `User-Agent`}
import org.http4s.util.{StringWriter, Writer}
import scala.annotation.tailrec
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.util.{Failure, Success}
import _root_.io.chrisdavenport.vault._

private final class Http1Connection[F[_]](
    val requestKey: RequestKey,
    protected override val executionContext: ExecutionContext,
    maxResponseLineSize: Int,
    maxHeaderLength: Int,
    maxChunkSize: Int,
    override val chunkBufferMaxSize: Int,
    parserMode: ParserMode,
    userAgent: Option[`User-Agent`]
)(implicit protected val F: ConcurrentEffect[F])
    extends Http1Stage[F]
    with BlazeConnection[F] {
  import org.http4s.client.blaze.Http1Connection._

  override def name: String = getClass.getName
  private val parser =
    new BlazeHttp1ClientParser(maxResponseLineSize, maxHeaderLength, maxChunkSize, parserMode)

  private val stageState = new AtomicReference[State](Idle(None))

  override def isClosed: Boolean =
    stageState.get match {
      case Error(_) => true
      case _ => false
    }

  override def isRecyclable: Boolean =
    stageState.get match {
      case Idle(_) => true
      case _ => false
    }

  override def shutdown(): Unit = stageShutdown()

  override def stageShutdown(): Unit = shutdownWithError(EOF)

  override protected def fatalError(t: Throwable, msg: String): Unit = {
    val realErr = t match {
      case _: TimeoutException => EOF
      case EOF => EOF
      case t =>
        logger.error(t)(s"Fatal Error: $msg")
        t
    }
    shutdownWithError(realErr)
  }

  @tailrec
  private def shutdownWithError(t: Throwable): Unit =
    stageState.get match {
      // If we have a real error, lets put it here.
      case st @ Error(EOF) if t != EOF =>
        if (!stageState.compareAndSet(st, Error(t))) shutdownWithError(t)
        else closePipeline(Some(t))

      case Error(_) => // NOOP: already shutdown

      case x =>
        if (!stageState.compareAndSet(x, Error(t))) shutdownWithError(t)
        else {
          val cmd = t match {
            case EOF => None
            case _ => Some(t)
          }
          closePipeline(cmd)
          super.stageShutdown()
        }
    }

  @tailrec
  def resetRead(): Unit = {
    val state = stageState.get()
    val nextState = state match {
      case i: Idle => Some(i)
      case ReadWrite => Some(Write)
      case Read => Some(Idle(Some(startIdleRead())))
      case _ => None
    }

    nextState match {
      case Some(n) => if (stageState.compareAndSet(state, n)) parser.reset() else resetRead()
      case None => ()
    }
  }

  @tailrec
  def resetWrite(): Unit = {
    val state = stageState.get()
    val nextState = state match {
      case i: Idle => Some(i)
      case ReadWrite => Some(Read)
      case Write => Some(Idle(Some(startIdleRead())))
      case _ => None
    }

    nextState match {
      case Some(n) => if (stageState.compareAndSet(state, n)) () else resetWrite()
      case None => ()
    }
  }

  // #4798 We read from the channel while the connection is idle, in order to receive an EOF when the connection gets closed.
  private def startIdleRead(): Future[ByteBuffer] = {
    val f = channelRead()
    f.onComplete {
      case Failure(t) => shutdownWithError(t)
      case _ =>
    }(executionContext)
    f
  }

  def runRequest(req: Request[F], idleTimeoutF: F[TimeoutException]): F[Response[F]] =
    F.defer[Response[F]] {
      stageState.get match {
        case i @ Idle(idleRead) =>
          if (stageState.compareAndSet(i, ReadWrite)) {
            logger.debug(s"Connection was idle. Running.")
            executeRequest(req, idleTimeoutF, idleRead)<commit_msg>avoid unnecessarily changing state of Http1Connection<commit_after>/*
 * Copyright 2014 http4s.org
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.http4s
package client
package blaze

import cats.effect._
import cats.effect.implicits._
import cats.syntax.all._
import fs2._
import java.nio.ByteBuffer
import java.util.concurrent.TimeoutException
import java.util.concurrent.atomic.AtomicReference
import org.http4s.{headers => H}
import org.http4s.Uri.{Authority, RegName}
import org.http4s.blaze.pipeline.Command.EOF
import org.http4s.blazecore.Http1Stage
import org.http4s.blazecore.util.Http1Writer
import org.http4s.headers.{Connection, Host, `Content-Length`, `User-Agent`}
import org.http4s.util.{StringWriter, Writer}
import scala.annotation.tailrec
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.util.{Failure, Success}
import _root_.io.chrisdavenport.vault._

private final class Http1Connection[F[_]](
    val requestKey: RequestKey,
    protected override val executionContext: ExecutionContext,
    maxResponseLineSize: Int,
    maxHeaderLength: Int,
    maxChunkSize: Int,
    override val chunkBufferMaxSize: Int,
    parserMode: ParserMode,
    userAgent: Option[`User-Agent`]
)(implicit protected val F: ConcurrentEffect[F])
    extends Http1Stage[F]
    with BlazeConnection[F] {
  import org.http4s.client.blaze.Http1Connection._

  override def name: String = getClass.getName
  private val parser =
    new BlazeHttp1ClientParser(maxResponseLineSize, maxHeaderLength, maxChunkSize, parserMode)

  private val stageState = new AtomicReference[State](Idle(None))

  override def isClosed: Boolean =
    stageState.get match {
      case Error(_) => true
      case _ => false
    }

  override def isRecyclable: Boolean =
    stageState.get match {
      case Idle(_) => true
      case _ => false
    }

  override def shutdown(): Unit = stageShutdown()

  override def stageShutdown(): Unit = shutdownWithError(EOF)

  override protected def fatalError(t: Throwable, msg: String): Unit = {
    val realErr = t match {
      case _: TimeoutException => EOF
      case EOF => EOF
      case t =>
        logger.error(t)(s"Fatal Error: $msg")
        t
    }
    shutdownWithError(realErr)
  }

  @tailrec
  private def shutdownWithError(t: Throwable): Unit =
    stageState.get match {
      // If we have a real error, lets put it here.
      case st @ Error(EOF) if t != EOF =>
        if (!stageState.compareAndSet(st, Error(t))) shutdownWithError(t)
        else closePipeline(Some(t))

      case Error(_) => // NOOP: already shutdown

      case x =>
        if (!stageState.compareAndSet(x, Error(t))) shutdownWithError(t)
        else {
          val cmd = t match {
            case EOF => None
            case _ => Some(t)
          }
          closePipeline(cmd)
          super.stageShutdown()
        }
    }

  @tailrec
  def resetRead(): Unit = {
    val state = stageState.get()
    val nextState = state match {
      case ReadWrite => Some(Write)
      case Read => Some(Idle(Some(startIdleRead())))
      case _ => None
    }

    nextState match {
      case Some(n) => if (stageState.compareAndSet(state, n)) parser.reset() else resetRead()
      case None => ()
    }
  }

  @tailrec
  def resetWrite(): Unit = {
    val state = stageState.get()
    val nextState = state match {
      case ReadWrite => Some(Read)
      case Write => Some(Idle(Some(startIdleRead())))
      case _ => None
    }

    nextState match {
      case Some(n) => if (stageState.compareAndSet(state, n)) () else resetWrite()
      case None => ()
    }
  }

  // #4798 We read from the channel while the connection is idle, in order to receive an EOF when the connection gets closed.
  private def startIdleRead(): Future[ByteBuffer] = {
    val f = channelRead()
    f.onComplete {
      case Failure(t) => shutdownWithError(t)
      case _ =>
    }(executionContext)
    f
  }

  def runRequest(req: Request[F], idleTimeoutF: F[TimeoutException]): F[Response[F]] =
    F.defer[Response[F]] {
      stageState.get match {
        case i @ Idle(idleRead) =>
          if (stageState.compareAndSet(i, ReadWrite)) {
            logger.debug(s"Connection was idle. Running.")
            executeRequest(req, idleTimeoutF, idleRead)
{'fraction_non_alphanumeric': 0.10288321924911845, 'fraction_numerical': 0.0061190624351794235, 'mean_word_length': 3.1174210076857385, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3837111', 'n_tokens_mistral': 2921, 'n_tokens_neox': 2848, 'n_words': 846}