Module: KafkaSyrup

Defined in:
lib/kafka_syrup.rb,
lib/kafka_syrup/utils.rb,
lib/kafka_syrup/broker.rb,
lib/kafka_syrup/errors.rb,
lib/kafka_syrup/version.rb,
lib/kafka_syrup/encoding.rb,
lib/kafka_syrup/protocol.rb,
lib/kafka_syrup/protocol/base.rb,
lib/kafka_syrup/topic_consumer.rb,
lib/kafka_syrup/topic_producer.rb,
lib/kafka_syrup/protocol/errors.rb,
lib/kafka_syrup/protocol/message.rb,
lib/kafka_syrup/protocol/request.rb,
lib/kafka_syrup/protocol/response.rb,
lib/kafka_syrup/partition_consumer.rb,
lib/kafka_syrup/protocol/message_set.rb,
lib/kafka_syrup/protocol/fetch_request.rb,
lib/kafka_syrup/protocol/fetch_response.rb,
lib/kafka_syrup/protocol/offset_request.rb,
lib/kafka_syrup/protocol/offset_response.rb,
lib/kafka_syrup/protocol/produce_request.rb,
lib/kafka_syrup/protocol/metadata_request.rb,
lib/kafka_syrup/protocol/produce_response.rb,
lib/kafka_syrup/protocol/metadata_response.rb

Defined Under Namespace

Modules: Encoding, KafkaResponseErrors, Protocol, Utils Classes: Broker, BrokerNotFound, Configuration, Error, KafkaResponseError, NoBrokers, NotRegistered, PartitionConsumer, PartitionNotFound, SocketReadError, TopicConsumer, TopicNotFound, TopicProducer

Constant Summary collapse

E =

Just to abbreviate the typing

Encoding
VERSION = 
"0.9.0"

Class Method Summary collapse

Class Method Details

.brokers ⇒ Object 



50
51
52
 
# File 'lib/kafka_syrup.rb', line 50

def brokers
  @brokers ||= config.brokers.split(',').map(&:strip).map{ |info| Broker.new(*info.split(':')) }
end

.config ⇒ Object 



46
47
48
 
# File 'lib/kafka_syrup.rb', line 46

def config
  @config ||= Configuration.new.with_defaults
end

.configure {|config| ... } ⇒ Object

Yields:



42
43
44
 
# File 'lib/kafka_syrup.rb', line 42

def configure
  yield config
end

.get_metadata(*topics) ⇒ Object

Raises:



54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/kafka_syrup.rb', line 54

def get_metadata(*topics)
  request = KafkaSyrup::Protocol::MetadataRequest.new(*topics)

  brokers.each do |broker|
    begin
      response = broker.send_request(request, close: true)
      return response
    rescue StandardError
    end
  end

  raise NoBrokers
end