Class: Phobos::Producer::ClassMethods::PublicAPI
- Inherits:
-
Object
- Object
- Phobos::Producer::ClassMethods::PublicAPI
- Defined in:
- lib/phobos/producer.rb
Constant Summary collapse
- NAMESPACE =
:phobos_producer_store
- ASYNC_PRODUCER_PARAMS =
%i(max_queue_size delivery_threshold delivery_interval).freeze
Instance Method Summary collapse
- #async_configs ⇒ Object
- #async_producer ⇒ Object
- #async_producer_shutdown ⇒ Object
- #async_publish(topic, payload, key = nil) ⇒ Object
- #async_publish_list(messages) ⇒ Object
-
#configure_kafka_client(kafka_client) ⇒ Object
This method configures the kafka client used with publish operations performed by the host class.
- #create_async_producer ⇒ Object
- #kafka_client ⇒ Object
- #publish(topic, payload, key = nil) ⇒ Object
- #publish_list(messages) ⇒ Object
- #regular_configs ⇒ Object
Instance Method Details
#async_configs ⇒ Object
109 110 111 |
# File 'lib/phobos/producer.rb', line 109 def async_configs Phobos.config.producer_hash end |
#async_producer ⇒ Object
86 87 88 |
# File 'lib/phobos/producer.rb', line 86 def async_producer producer_store[:async_producer] end |
#async_producer_shutdown ⇒ Object
99 100 101 102 103 |
# File 'lib/phobos/producer.rb', line 99 def async_producer_shutdown async_producer&. async_producer&.shutdown producer_store[:async_producer] = nil end |
#async_publish(topic, payload, key = nil) ⇒ Object
90 91 92 |
# File 'lib/phobos/producer.rb', line 90 def async_publish(topic, payload, key = nil) async_publish_list([{ topic: topic, payload: payload, key: key }]) end |
#async_publish_list(messages) ⇒ Object
94 95 96 97 |
# File 'lib/phobos/producer.rb', line 94 def async_publish_list() producer = async_producer || create_async_producer (producer, ) end |
#configure_kafka_client(kafka_client) ⇒ Object
This method configures the kafka client used with publish operations performed by the host class
59 60 61 62 |
# File 'lib/phobos/producer.rb', line 59 def configure_kafka_client(kafka_client) async_producer_shutdown producer_store[:kafka_client] = kafka_client end |
#create_async_producer ⇒ Object
80 81 82 83 84 |
# File 'lib/phobos/producer.rb', line 80 def create_async_producer client = kafka_client || configure_kafka_client(Phobos.create_kafka_client) async_producer = client.async_producer(async_configs) producer_store[:async_producer] = async_producer end |
#kafka_client ⇒ Object
64 65 66 |
# File 'lib/phobos/producer.rb', line 64 def kafka_client producer_store[:kafka_client] end |
#publish(topic, payload, key = nil) ⇒ Object
68 69 70 |
# File 'lib/phobos/producer.rb', line 68 def publish(topic, payload, key = nil) publish_list([{ topic: topic, payload: payload, key: key }]) end |
#publish_list(messages) ⇒ Object
72 73 74 75 76 77 78 |
# File 'lib/phobos/producer.rb', line 72 def publish_list() client = kafka_client || configure_kafka_client(Phobos.create_kafka_client) producer = client.producer(regular_configs) (producer, ) ensure producer&.shutdown end |
#regular_configs ⇒ Object
105 106 107 |
# File 'lib/phobos/producer.rb', line 105 def regular_configs Phobos.config.producer_hash.reject { |k, _| ASYNC_PRODUCER_PARAMS.include?(k)} end |