<commit_before>package io.scalac.amqp.impl

import java.util.UUID
import java.util.concurrent.atomic.AtomicLong

import com.rabbitmq.client._

import io.scalac.amqp.Delivery

import org.reactivestreams.{Subscription, Subscriber}


private[amqp] class QueueSubscription(channel: Channel, queue: String, subscriber: Subscriber[_ >: Delivery])
  extends DefaultConsumer(channel) with Subscription {
  val tag = UUID.randomUUID().toString
  val demand = new AtomicLong()

  override def handleCancel(consumerTag: String) = try {
    subscriber.onComplete()
  } catch {
    case exception: Exception =>
      subscriber.onError(new IllegalStateException("Rule 2.13", exception))
  }

  override def handleShutdownSignal(consumerTag: String, sig: ShutdownSignalException) = sig match {
    case sig if !sig.isInitiatedByApplication => subscriber.onError(sig)
    case _ => // shutdown initiated by us
  }

  def prefetch(demand: Long): Unit = demand match {
    case demand if demand < Int.MaxValue => channel.basicQos(demand.toInt, true)
    case Long.MaxValue => channel.basicQos(0, true) // 3.17: effectively unbounded
    case _ => channel.basicQos(Int.MaxValue, true)
  }

  override def handleDelivery(consumerTag: String,
                              envelope: Envelope,
                              properties: AMQP.BasicProperties,
                              body: Array[Byte]) = {
    val delivery = Conversions.toDelivery(envelope, properties, body)
    demand.decrementAndGet() match {
      case 0 => getChannel.basicCancel(tag)
      case demand => prefetch(demand)<commit_msg>Extend ISE message for rule 2.13<commit_after>package io.scalac.amqp.impl

import java.util.UUID
import java.util.concurrent.atomic.AtomicLong

import com.rabbitmq.client._

import io.scalac.amqp.Delivery

import org.reactivestreams.{Subscription, Subscriber}


private[amqp] class QueueSubscription(channel: Channel, queue: String, subscriber: Subscriber[_ >: Delivery])
  extends DefaultConsumer(channel) with Subscription {
  val tag = UUID.randomUUID().toString
  val demand = new AtomicLong()

  override def handleCancel(consumerTag: String) = try {
    subscriber.onComplete()
  } catch {
    case exception: Exception =>
      subscriber.onError(new IllegalStateException("Rule 2.13 violation", exception))
  }

  override def handleShutdownSignal(consumerTag: String, sig: ShutdownSignalException) = sig match {
    case sig if !sig.isInitiatedByApplication => subscriber.onError(sig)
    case _ => // shutdown initiated by us
  }

  def prefetch(demand: Long): Unit = demand match {
    case demand if demand < Int.MaxValue => channel.basicQos(demand.toInt, true)
    case Long.MaxValue => channel.basicQos(0, true) // 3.17: effectively unbounded
    case _ => channel.basicQos(Int.MaxValue, true)
  }

  override def handleDelivery(consumerTag: String,
                              envelope: Envelope,
                              properties: AMQP.BasicProperties,
                              body: Array[Byte]) = {
    val delivery = Conversions.toDelivery(envelope, properties, body)
    demand.decrementAndGet() match {
      case 0 => getChannel.basicCancel(tag)
      case demand => prefetch(demand)
{'fraction_non_alphanumeric': 0.09753817388594578, 'fraction_numerical': 0.005920847616079776, 'mean_word_length': 3.4337016574585637, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 21, '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': '1789429', 'n_tokens_mistral': 943, 'n_tokens_neox': 899, 'n_words': 227}