<commit_before>require 'hermann'
require 'hermann/result'


if RUBY_PLATFORM == "java"
  require 'hermann/provider/java_producer'
else
  require 'hermann_lib'
end

module Hermann
  class Producer
    attr_reader :topic, :brokers, :internal, :children

    # Initialize a producer object with a default topic and broker list
    #
    # @param [String] topic The default topic to use for pushing messages
    # @param [Array] brokers An array of "host:port" strings for the brokers
    def initialize(topic, brokers)
      @topic = topic
      @brokers = brokers
      if RUBY_PLATFORM == "java"
        @internal = Hermann::Provider::JavaProducer.new(brokers)
      else
        @internal = Hermann::Lib::Producer.new(brokers)
      end
      # We're tracking children so we can make sure that at Producer exit we
      # make a reasonable attempt to clean up outstanding result objects
      @children = []
    end

    # @return [Boolean] True if our underlying producer object thinks it's
    #   connected to a Kafka broker
    def connected?
      return @internal.connected?
    end

    # @return [Boolean] True if the underlying producer object has errored
    def errored?
      return @internal.errored?
    end

    def connect(timeout=0)
      return @internal.connect(timeout * 1000)
    end

    # Push a value onto the Kafka topic passed to this +Producer+
    #
    # @param [Object] value A single object to push
    # @param [Hash] opts to pass to push method
    # @option opts [String] :topic The topic to push messages to
    #
    # @return [Hermann::Result] A future-like object which will store the
    #   result from the broker
    def push(value, opts={})
      topic = opts[:topic] || @topic
      result = nil

      if value.kind_of? Array
        return value.map { |e| self.push(e) }
      end

      if RUBY_PLATFORM == "java"
        result = @internal.push_single(value, topic)
        @children << result
        # Reaping children on the push just to make sure that it does get<commit_msg>Avoid adding non-existent children to our internal list.<commit_after>require 'hermann'
require 'hermann/result'


if RUBY_PLATFORM == "java"
  require 'hermann/provider/java_producer'
else
  require 'hermann_lib'
end

module Hermann
  class Producer
    attr_reader :topic, :brokers, :internal, :children

    # Initialize a producer object with a default topic and broker list
    #
    # @param [String] topic The default topic to use for pushing messages
    # @param [Array] brokers An array of "host:port" strings for the brokers
    def initialize(topic, brokers)
      @topic = topic
      @brokers = brokers
      if RUBY_PLATFORM == "java"
        @internal = Hermann::Provider::JavaProducer.new(brokers)
      else
        @internal = Hermann::Lib::Producer.new(brokers)
      end
      # We're tracking children so we can make sure that at Producer exit we
      # make a reasonable attempt to clean up outstanding result objects
      @children = []
    end

    # @return [Boolean] True if our underlying producer object thinks it's
    #   connected to a Kafka broker
    def connected?
      return @internal.connected?
    end

    # @return [Boolean] True if the underlying producer object has errored
    def errored?
      return @internal.errored?
    end

    def connect(timeout=0)
      return @internal.connect(timeout * 1000)
    end

    # Push a value onto the Kafka topic passed to this +Producer+
    #
    # @param [Object] value A single object to push
    # @param [Hash] opts to pass to push method
    # @option opts [String] :topic The topic to push messages to
    #
    # @return [Hermann::Result] A future-like object which will store the
    #   result from the broker
    def push(value, opts={})
      topic = opts[:topic] || @topic
      result = nil

      if value.kind_of? Array
        return value.map { |e| self.push(e) }
      end

      if RUBY_PLATFORM == "java"
        result = @internal.push_single(value, topic)
        unless result.nil?
          @children << result
        end
        # Reaping children on the push just to make sure that it does get
{'fraction_non_alphanumeric': 0.08027079303675048, 'fraction_numerical': 0.0024177949709864605, 'mean_word_length': 2.7540834845735027, 'pattern_counts': {'":': 0, '<': 7, '<?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': '2847459', 'n_tokens_mistral': 1208, 'n_tokens_neox': 1154, 'n_words': 457}