Class: LogStashLogger::Device::Kafka

Inherits:
Connectable show all
Defined in:
lib/logstash-logger/device/kafka.rb

Defined Under Namespace

Classes: TLSConfiguration

Instance Attribute Summary collapse

Attributes inherited from Connectable

#buffer_logger

Attributes inherited from Base

#error_logger, #io, #sync

Instance Method Summary collapse

Methods inherited from Connectable

#close, #connected?, #flush, #on_full_buffer_receive, #reconnect, #reset, #to_io, #with_connection, #write

Methods included from Buffer

#buffer_flush, #buffer_full?, #buffer_initialize, #buffer_receive, #reset_buffer

Methods inherited from Base

#close, #flush, #reset, #to_io, #unrecoverable_error?, #write

Constructor Details

#initialize(opts = {}, kafka_tls_configurator = TLSConfiguration) ⇒ Kafka

Returns a new instance of Kafka.



50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/logstash-logger/device/kafka.rb', line 50

def initialize(opts = {}, kafka_tls_configurator = TLSConfiguration)
  require 'ruby-kafka'
  super(opts)

  opts = normalize_uri_options(opts)
  @client_id = opts[:client_id]
  @topic = normalize_topic(opts[:topic])
  @buffer_group = @topic
  @kafka_tls_configurator = kafka_tls_configurator
  @brokers = make_brokers_array(opts[:brokers])
  raise_no_brokers_set! if @brokers.empty?
  make_cert_bundle(opts)
end

Instance Attribute Details

#brokers ⇒ Object (readonly)

Returns the value of attribute brokers.



47
48
49
 
# File 'lib/logstash-logger/device/kafka.rb', line 47

def brokers
  @brokers
end

#cert_bundle ⇒ Object (readonly)

Returns the value of attribute cert_bundle.



47
48
49
 
# File 'lib/logstash-logger/device/kafka.rb', line 47

def cert_bundle
  @cert_bundle
end

#client_id ⇒ Object (readonly)

Returns the value of attribute client_id.



47
48
49
 
# File 'lib/logstash-logger/device/kafka.rb', line 47

def client_id
  @client_id
end

#kafka_tls_configurator ⇒ Object (readonly)

Returns the value of attribute kafka_tls_configurator.



47
48
49
 
# File 'lib/logstash-logger/device/kafka.rb', line 47

def kafka_tls_configurator
  @kafka_tls_configurator
end

#topic ⇒ Object (readonly)

Returns the value of attribute topic.



47
48
49
 
# File 'lib/logstash-logger/device/kafka.rb', line 47

def topic
  @topic
end

Instance Method Details

#connect ⇒ Object 



68
69
70
 
# File 'lib/logstash-logger/device/kafka.rb', line 68

def connect
  @io = connection
end

#connection ⇒ Object 



64
65
66
 
# File 'lib/logstash-logger/device/kafka.rb', line 64

def connection
  @io ||= ::Kafka.new(**kafka_client_connection_hash)
end

#write_batch(messages, topic = nil) ⇒ Object 



79
80
81
82
83
84
 
# File 'lib/logstash-logger/device/kafka.rb', line 79

def write_batch(messages, topic=nil)
  topic ||= @topic
  write_messages_to_broker_and_deliver do |producer|
    messages.each {|msg| producer.produce(msg, topic: topic) }
  end
end

#write_one(message, topic = nil) ⇒ Object 



72
73
74
75
76
77
 
# File 'lib/logstash-logger/device/kafka.rb', line 72

def write_one(message, topic=nil)
  topic ||= @topic
  write_messages_to_broker_and_deliver do |producer|
    producer.produce(message, topic: topic)
  end
end