<commit_before>module Rdkafka
  # A producer for Kafka messages. To create a producer set up a {Config} and call {Config#producer producer} on that.
  class Producer
    # @private
    # Returns the current delivery callback, by default this is nil.
    #
    # @return [Proc, nil]
    attr_reader :delivery_callback

    # @private
    def initialize(native_kafka)
      @closing = false
      @native_kafka = native_kafka

      # Makes sure, that the producer gets closed before it gets GCed by Ruby
      ObjectSpace.define_finalizer(self, proc { close })

      # Start thread to poll client for delivery callbacks
      @polling_thread = Thread.new do
        loop do
          Rdkafka::Bindings.rd_kafka_poll(@native_kafka, 250)
          # Exit thread if closing and the poll queue is empty
          if @closing && Rdkafka::Bindings.rd_kafka_outq_len(@native_kafka) == 0
            break
          end
        end
      end
      @polling_thread.abort_on_exception = true
    end

    # Set a callback that will be called every time a message is successfully produced.
    # The callback is called with a {DeliveryReport}
    #
    # @param callback [Proc, #call] The callback
    #
    # @return [nil]
    def delivery_callback=(callback)
      raise TypeError.new("Callback has to be callable") unless callback.respond_to?(:call)
      @delivery_callback = callback
    end

    # Close this producer and wait for the internal poll queue to empty.
    def close
      return unless @native_kafka

      # Indicate to polling thread that we're closing
      @closing = true
      # Wait for the polling thread to finish up
      @polling_thread.join
      Rdkafka::Bindings.rd_kafka_destroy(@native_kafka)
      @native_kafka = nil
    end

    # Partition count for a given topic.
    # NOTE: If 'allow.auto.create.topics' is set to true in the broker, the topic will be auto-created after returning nil.
    #
    # @param topic [String] The topic name.
    #
    # @return partition count [Integer,nil]<commit_msg>Use id instead of object to bind the the finalizer for a producer<commit_after>require "securerandom"

module Rdkafka
  # A producer for Kafka messages. To create a producer set up a {Config} and call {Config#producer producer} on that.
  class Producer
    # @private
    # Returns the current delivery callback, by default this is nil.
    #
    # @return [Proc, nil]
    attr_reader :delivery_callback

    # @private
    def initialize(native_kafka)
      @id = SecureRandom.uuid
      @closing = false
      @native_kafka = native_kafka

      # Makes sure, that the producer gets closed before it gets GCed by Ruby
      ObjectSpace.define_finalizer(@id, proc { close })

      # Start thread to poll client for delivery callbacks
      @polling_thread = Thread.new do
        loop do
          Rdkafka::Bindings.rd_kafka_poll(@native_kafka, 250)
          # Exit thread if closing and the poll queue is empty
          if @closing && Rdkafka::Bindings.rd_kafka_outq_len(@native_kafka) == 0
            break
          end
        end
      end
      @polling_thread.abort_on_exception = true
    end

    # Set a callback that will be called every time a message is successfully produced.
    # The callback is called with a {DeliveryReport}
    #
    # @param callback [Proc, #call] The callback
    #
    # @return [nil]
    def delivery_callback=(callback)
      raise TypeError.new("Callback has to be callable") unless callback.respond_to?(:call)
      @delivery_callback = callback
    end

    # Close this producer and wait for the internal poll queue to empty.
    def close
      ObjectSpace.undefine_finalizer(@id)

      return unless @native_kafka

      # Indicate to polling thread that we're closing
      @closing = true
      # Wait for the polling thread to finish up
      @polling_thread.join
      Rdkafka::Bindings.rd_kafka_destroy(@native_kafka)
      @native_kafka = nil
    end

    # Partition count for a given topic.
    # NOTE: If 'allow.auto.create.topics' is set to true in the broker, the topic will be auto-created after returning nil.
    #
    # @param topic [String] The topic name.
    #
    # @return partition count [Integer,nil]
{'fraction_non_alphanumeric': 0.06850618458610847, 'fraction_numerical': 0.0019029495718363464, 'mean_word_length': 2.711385701676964, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '886033', 'n_tokens_mistral': 1216, 'n_tokens_neox': 1176, 'n_words': 442}