<commit_before>package io.scalac.amqp.impl

import scala.concurrent.stm.Ref

import com.rabbitmq.client.{ShutdownSignalException, ShutdownListener, Connection}

import io.scalac.amqp.Delivery

import org.reactivestreams.{Subscriber, Publisher}


private[amqp] class QueuePublisher(connection: Connection, queue: String, prefetch: Int = 20)
  extends Publisher[Delivery] {
  val subscribers = Ref(Set[Subscriber[_ >: Delivery]]())

  override def subscribe(subscriber: Subscriber[_ >: Delivery]) =
    subscribers.single.getAndTransform(_ + subscriber) match {
      case ss if ss.contains(subscriber) ⇒
        throw new IllegalStateException(s"1.10: Subscriber=$subscriber is already subscribed to this publisher.")
      case _                             ⇒ try {
        val channel = connection.createChannel()
        channel.addShutdownListener(newShutdownListener(subscriber))

        val subscription = new QueueSubscription(channel, subscriber)
        subscriber.onSubscribe(subscription)

        channel.basicQos(prefetch)
        channel.basicConsume(queue, false, subscription)
      } catch {
        case exception: Exception ⇒ subscriber.onError(exception)
      }
    }

  def newShutdownListener(subscriber: Subscriber[_ >: Delivery]) = new ShutdownListener {
    override def shutdownCompleted(cause: ShutdownSignalException) =
      subscribers.single.transform(_ - subscriber)
  }
}
<commit_msg>Check if prefetch > 0<commit_after>package io.scalac.amqp.impl

import scala.concurrent.stm.Ref

import com.rabbitmq.client.{ShutdownSignalException, ShutdownListener, Connection}

import io.scalac.amqp.Delivery

import org.reactivestreams.{Subscriber, Publisher}


private[amqp] class QueuePublisher(connection: Connection, queue: String, prefetch: Int = 20)
  extends Publisher[Delivery] {
  require(prefetch > 0, "prefetch <= 0")

  val subscribers = Ref(Set[Subscriber[_ >: Delivery]]())

  override def subscribe(subscriber: Subscriber[_ >: Delivery]) =
    subscribers.single.getAndTransform(_ + subscriber) match {
      case ss if ss.contains(subscriber) ⇒
        throw new IllegalStateException(s"1.10: Subscriber=$subscriber is already subscribed to this publisher.")
      case _                             ⇒ try {
        val channel = connection.createChannel()
        channel.addShutdownListener(newShutdownListener(subscriber))

        val subscription = new QueueSubscription(channel, subscriber)
        subscriber.onSubscribe(subscription)

        channel.basicQos(prefetch)
        channel.basicConsume(queue, false, subscription)
      } catch {
        case exception: Exception ⇒ subscriber.onError(exception)
      }
    }

  def newShutdownListener(subscriber: Subscriber[_ >: Delivery]) = new ShutdownListener {
    override def shutdownCompleted(cause: ShutdownSignalException) =
      subscribers.single.transform(_ - subscriber)
  }
}

{'fraction_non_alphanumeric': 0.0932085932085932, 'fraction_numerical': 0.0045045045045045045, 'mean_word_length': 3.8439597315436242, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 11, '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': '976741', 'n_tokens_mistral': 835, 'n_tokens_neox': 796, 'n_words': 179}