<commit_before># encoding: utf-8

module Bunny

  # Queues store and forward messages. Queues can be configured in the server or created at runtime.
  # Queues must be attached to at least one exchange in order to receive messages from publishers.
  class Queue09 < Qrack::Queue

    def initialize(client, name, opts = {})
      # check connection to server
      raise Bunny::ConnectionError, 'Not connected to server' if client.status == :not_connected

      @client = client
      @opts   = opts
      @delivery_tag = nil

      # Queues without a given name are named by the server and are generally
      # bound to the process that created them.
      if !name
        opts = {
          :passive => false,
          :durable => false,
          :exclusive => true,
          :auto_delete => true,
          :deprecated_ticket => 0
        }.merge(opts)
      end

      # ignore the :nowait option if passed, otherwise program will hang waiting for a
      # response that will not be sent by the server
      opts.delete(:nowait)

      opts = { :queue => name || '', :nowait => false, :deprecated_ticket => 0 }.merge(opts)

      client.send_frame(Qrack::Protocol09::Queue::Declare.new(opts))

      method = client.next_method

      client.check_response(method, Qrack::Protocol09::Queue::DeclareOk, "Error declaring queue #{name}")

      @name = method.queue
      client.queues[@name] = self
    end

    # Acknowledges one or more messages delivered via the _Deliver_ or _Get_-_Ok_ methods. The client can
    # ask to confirm a single message or a set of messages up to and including a specific message.
    #
    # @option opts [String] :delivery_tag
    #
    # @option opts [Boolean] :multiple (false)
    #   If set to @true@, the delivery tag is treated as "up to and including",
    #   so that the client can acknowledge multiple messages with a single method.
    #   If set to @false@, the delivery tag refers to a single message.
    #   If the multiple field is @true@, and the delivery tag is zero,
    #   tells the server to acknowledge all outstanding messages.
    def ack(opts = {})
      # Set delivery tag
      if delivery_tag.nil? and opts[:delivery_tag].nil?
        raise Bunny::AcknowledgementError, "No delivery tag received"
      else
        self.delivery_tag = opts[:delivery_tag] if delivery_tag.nil?
      end

      opts = {:delivery_tag => delivery_tag, :multiple => false}.merge(opts)

      client.send_frame(Qrack::Protocol09::Basic::Ack.new(opts))

      # reset delivery tag
      self.delivery_tag = nil
    end

    # Binds a queue to an exchange. Until a queue is bound it won't receive
    # any messages. Queues are bound to the direct exchange '' by default.
    # If error occurs, a {Bunny::ProtocolError} is raised.
    #
    # @option opts [String] :key
    #   Specifies the routing key for the binding. The routing key is used
    #   for routing messages depending on the exchange configuration.
    #
    # @option opts [Boolean] :nowait (false)
    #   Ignored by Bunny, always @false@.
    #
    # @return [Symbol] @:bind_ok@ if successful.
    def bind(exchange, opts = {})
      exchange = exchange.respond_to?(:name) ? exchange.name : exchange

      # ignore the :nowait option if passed, otherwise program will hang waiting for a
      # response that will not be sent by the server
      opts.delete(:nowait)

      opts = {
        :queue => name,
        :exchange => exchange,
        :routing_key => opts.delete(:key),
        :nowait => false,
        :deprecated_ticket => 0
      }.merge(opts)

      client.send_frame(Qrack::Protocol09::Queue::Bind.new(opts))

      method = client.next_method

      client.check_response(method, Qrack::Protocol09::Queue::BindOk, "Error binding queue: #{name} to exchange: #{exchange}")

      # return message
      :bind_ok
    end

    # Requests that a queue is deleted from broker/server. When a queue is deleted any pending messages
    # are sent to a dead-letter queue if this is defined in the server configuration. Removes reference
    # from queues if successful. If an error occurs raises _Bunny_::_ProtocolError_.
    #
    # @option opts [Boolean] :if_unused (false)
    #   If set to @true@, the server will only delete the queue if it has no consumers.
    #   If the queue has consumers the server does not delete it but raises a channel exception instead.
    #
    # @option opts [Boolean] :if_empty (false)
    #   If set to @true@, the server will only delete the queue if it has no messages.
    #   If the queue is not empty the server raises a channel exception.
    #
    # @option opts [Boolean] :nowait (false)
    #   Ignored by Bunny, always @false@.
    #
    # @return [Symbol] @:delete_ok@ if successful
    def delete(opts = {})
      # ignore the :nowait option if passed, otherwise program will hang waiting for a
      # response that will not be sent by the server
      opts.delete(:nowait)

      opts = { :queue => name, :nowait => false, :deprecated_ticket => 0 }.merge(opts)

      client.send_frame(Qrack::Protocol09::Queue::Delete.new(opts))

      method = client.next_method

      client.check_response(method, Qrack::Protocol09::Queue::DeleteOk, "Error deleting queue #{name}")

      client.queues.delete(name)

      # return confirmation
      :delete_ok
    end

    # Gets a message from a queue in a synchronous way. If error occurs, raises _Bunny_::_ProtocolError_.
    #
    # @option opts [Boolean] :ack (false)
    #   If set to @false@, the server does not expect an acknowledgement message
    #   from the client. If set to @true@, the server expects an acknowledgement
    #   message from the client and will re-queue the message if it does not receive
    #   one within a time specified by the server.
    #
    # @return [Hash] Hash with @:header@, @:payload@ and @:delivery_details@ keys. @:delivery_details@ is a hash @:consumer_tag@, @:delivery_tag@, @:redelivered@, @:exchange@ and @:routing_key@. If the queue is empty the returned hash will contain: @{:header => nil, :payload => :queue_empty, :delivery_details => nil}@. N.B. If a block is provided then the hash will be passed into the block and the return value will be nil.
    def pop(opts = {}, &blk)

      # do we want to have to provide an acknowledgement?
      ack = opts.delete(:ack)

      opts = {
        :queue => name,
        :consumer_tag => name,
        :no_ack => !ack,
        :nowait => true,
        :deprecated_ticket => 0
      }.merge(opts)

      client.send_frame(Qrack::Protocol09::Basic::Get.new(opts))

      method = client.next_method

      if method.is_a?(Qrack::Protocol09::Basic::GetEmpty) then
        queue_empty = true
      elsif !method.is_a?(Qrack::Protocol09::Basic::GetOk)
        raise Bunny::ProtocolError, "Error getting message from queue #{name}"
      end

      if !queue_empty
        # get delivery tag to use for acknowledge
        self.delivery_tag = method.delivery_tag if ack

        header = client.next_payload

        # If maximum frame size is smaller than message payload body then message
        # will have a message header and several message bodies
        msg = ''
        while msg.length < header.size
          msg << client.next_payload
        end

        msg_hash = {:header => header, :payload => msg, :delivery_details => method.arguments}

      else
        msg_hash = {:header => nil, :payload => :queue_empty, :delivery_details => nil}<commit_msg>Do not remove :ack from option hash.<commit_after># encoding: utf-8

module Bunny

  # Queues store and forward messages. Queues can be configured in the server or created at runtime.
  # Queues must be attached to at least one exchange in order to receive messages from publishers.
  class Queue09 < Qrack::Queue

    def initialize(client, name, opts = {})
      # check connection to server
      raise Bunny::ConnectionError, 'Not connected to server' if client.status == :not_connected

      @client = client
      @opts   = opts
      @delivery_tag = nil

      # Queues without a given name are named by the server and are generally
      # bound to the process that created them.
      if !name
        opts = {
          :passive => false,
          :durable => false,
          :exclusive => true,
          :auto_delete => true,
          :deprecated_ticket => 0
        }.merge(opts)
      end

      # ignore the :nowait option if passed, otherwise program will hang waiting for a
      # response that will not be sent by the server
      opts.delete(:nowait)

      opts = { :queue => name || '', :nowait => false, :deprecated_ticket => 0 }.merge(opts)

      client.send_frame(Qrack::Protocol09::Queue::Declare.new(opts))

      method = client.next_method

      client.check_response(method, Qrack::Protocol09::Queue::DeclareOk, "Error declaring queue #{name}")

      @name = method.queue
      client.queues[@name] = self
    end

    # Acknowledges one or more messages delivered via the _Deliver_ or _Get_-_Ok_ methods. The client can
    # ask to confirm a single message or a set of messages up to and including a specific message.
    #
    # @option opts [String] :delivery_tag
    #
    # @option opts [Boolean] :multiple (false)
    #   If set to @true@, the delivery tag is treated as "up to and including",
    #   so that the client can acknowledge multiple messages with a single method.
    #   If set to @false@, the delivery tag refers to a single message.
    #   If the multiple field is @true@, and the delivery tag is zero,
    #   tells the server to acknowledge all outstanding messages.
    def ack(opts = {})
      # Set delivery tag
      if delivery_tag.nil? and opts[:delivery_tag].nil?
        raise Bunny::AcknowledgementError, "No delivery tag received"
      else
        self.delivery_tag = opts[:delivery_tag] if delivery_tag.nil?
      end

      opts = {:delivery_tag => delivery_tag, :multiple => false}.merge(opts)

      client.send_frame(Qrack::Protocol09::Basic::Ack.new(opts))

      # reset delivery tag
      self.delivery_tag = nil
    end

    # Binds a queue to an exchange. Until a queue is bound it won't receive
    # any messages. Queues are bound to the direct exchange '' by default.
    # If error occurs, a {Bunny::ProtocolError} is raised.
    #
    # @option opts [String] :key
    #   Specifies the routing key for the binding. The routing key is used
    #   for routing messages depending on the exchange configuration.
    #
    # @option opts [Boolean] :nowait (false)
    #   Ignored by Bunny, always @false@.
    #
    # @return [Symbol] @:bind_ok@ if successful.
    def bind(exchange, opts = {})
      exchange = exchange.respond_to?(:name) ? exchange.name : exchange

      # ignore the :nowait option if passed, otherwise program will hang waiting for a
      # response that will not be sent by the server
      opts.delete(:nowait)

      opts = {
        :queue => name,
        :exchange => exchange,
        :routing_key => opts.delete(:key),
        :nowait => false,
        :deprecated_ticket => 0
      }.merge(opts)

      client.send_frame(Qrack::Protocol09::Queue::Bind.new(opts))

      method = client.next_method

      client.check_response(method, Qrack::Protocol09::Queue::BindOk, "Error binding queue: #{name} to exchange: #{exchange}")

      # return message
      :bind_ok
    end

    # Requests that a queue is deleted from broker/server. When a queue is deleted any pending messages
    # are sent to a dead-letter queue if this is defined in the server configuration. Removes reference
    # from queues if successful. If an error occurs raises _Bunny_::_ProtocolError_.
    #
    # @option opts [Boolean] :if_unused (false)
    #   If set to @true@, the server will only delete the queue if it has no consumers.
    #   If the queue has consumers the server does not delete it but raises a channel exception instead.
    #
    # @option opts [Boolean] :if_empty (false)
    #   If set to @true@, the server will only delete the queue if it has no messages.
    #   If the queue is not empty the server raises a channel exception.
    #
    # @option opts [Boolean] :nowait (false)
    #   Ignored by Bunny, always @false@.
    #
    # @return [Symbol] @:delete_ok@ if successful
    def delete(opts = {})
      # ignore the :nowait option if passed, otherwise program will hang waiting for a
      # response that will not be sent by the server
      opts.delete(:nowait)

      opts = { :queue => name, :nowait => false, :deprecated_ticket => 0 }.merge(opts)

      client.send_frame(Qrack::Protocol09::Queue::Delete.new(opts))

      method = client.next_method

      client.check_response(method, Qrack::Protocol09::Queue::DeleteOk, "Error deleting queue #{name}")

      client.queues.delete(name)

      # return confirmation
      :delete_ok
    end

    # Gets a message from a queue in a synchronous way. If error occurs, raises _Bunny_::_ProtocolError_.
    #
    # @option opts [Boolean] :ack (false)
    #   If set to @false@, the server does not expect an acknowledgement message
    #   from the client. If set to @true@, the server expects an acknowledgement
    #   message from the client and will re-queue the message if it does not receive
    #   one within a time specified by the server.
    #
    # @return [Hash] Hash with @:header@, @:payload@ and @:delivery_details@ keys. @:delivery_details@ is a hash @:consumer_tag@, @:delivery_tag@, @:redelivered@, @:exchange@ and @:routing_key@. If the queue is empty the returned hash will contain: @{:header => nil, :payload => :queue_empty, :delivery_details => nil}@. N.B. If a block is provided then the hash will be passed into the block and the return value will be nil.
    def pop(opts = {}, &blk)
      opts = {
        :queue => name,
        :consumer_tag => name,
        :no_ack => !opts[:ack],
        :nowait => true,
        :deprecated_ticket => 0
      }.merge(opts)

      client.send_frame(Qrack::Protocol09::Basic::Get.new(opts))

      method = client.next_method

      if method.is_a?(Qrack::Protocol09::Basic::GetEmpty) then
        queue_empty = true
      elsif !method.is_a?(Qrack::Protocol09::Basic::GetOk)
        raise Bunny::ProtocolError, "Error getting message from queue #{name}"
      end

      if !queue_empty
        # get delivery tag to use for acknowledge
        self.delivery_tag = method.delivery_tag if opts[:ack]

        header = client.next_payload

        # If maximum frame size is smaller than message payload body then message
        # will have a message header and several message bodies
        msg = ''
        while msg.length < header.size
          msg << client.next_payload
        end

        msg_hash = {:header => header, :payload => msg, :delivery_details => method.arguments}

      else
        msg_hash = {:header => nil, :payload => :queue_empty, :delivery_details => nil}
{'fraction_non_alphanumeric': 0.09508152720928673, 'fraction_numerical': 0.0037576326914044154, 'mean_word_length': 2.8020408163265307, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 67, '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': '1029749', 'n_tokens_mistral': 4324, 'n_tokens_neox': 4160, 'n_words': 1606}