<s> import logging <EOL> import time <EOL> from pykafka . exceptions import RdKafkaStoppedException , ProducerStoppedException <EOL> from pykafka . producer import Producer , CompressionType , random_partitioner <EOL> from pykafka . protocol import Message <EOL> from pykafka . utils . compat import get_bytes <EOL> from . import _rd_kafka <EOL> log = logging . getLogger ( __name__ ) <EOL> class RdKafkaProducer ( Producer ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> cluster , <EOL> topic , <EOL> partitioner = random_partitioner , <EOL> compression = CompressionType . NONE , <EOL> max_retries = <NUM_LIT:3> , <EOL> retry_backoff_ms = <NUM_LIT:100> , <EOL> required_acks = <NUM_LIT:1> , <EOL> ack_timeout_ms = <NUM_LIT:10> * <NUM_LIT:1000> , <EOL> max_queued_messages = <NUM_LIT> , <EOL> min_queued_messages = <NUM_LIT> , <EOL> linger_ms = <NUM_LIT:5> * <NUM_LIT:1000> , <EOL> block_on_queue_full = True , <EOL> sync = False , <EOL> delivery_reports = False , <EOL> auto_start = True ) : <EOL> callargs = { k : v for k , v in vars ( ) . items ( ) <EOL> if k not in ( "<STR_LIT>" , "<STR_LIT>" ) } <EOL> self . _rdk_producer = None <EOL> self . _poller_thread = None <EOL> self . _stop_poller_thread = cluster . handler . Event ( ) <EOL> super ( RdKafkaProducer , self ) . __init__ ( ** callargs ) <EOL> def start ( self ) : <EOL> if not self . _running : <EOL> brokers = b'<STR_LIT:U+002C>' . join ( b . host + b"<STR_LIT::>" + get_bytes ( str ( b . port ) ) <EOL> for b in self . _cluster . brokers . values ( ) ) <EOL> conf , topic_conf = self . _mk_rdkafka_config_lists ( ) <EOL> self . _rdk_producer = _rd_kafka . Producer ( ) <EOL> self . _rdk_producer . configure ( conf = conf ) <EOL> self . _rdk_producer . configure ( topic_conf = topic_conf ) <EOL> self . _rdk_producer . start ( <EOL> brokers , self . _topic . name , self . _delivery_reports . put ) <EOL> self . _running = True <EOL> def poll ( rdk_handle , stop_event ) : <EOL> while not stop_event . is_set ( ) or rdk_handle . outq_len ( ) > <NUM_LIT:0> : <EOL> rdk_handle . poll ( timeout_ms = <NUM_LIT:1000> ) <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . _stop_poller_thread . clear ( ) <EOL> self . _poller_thread = self . _cluster . handler . spawn ( <EOL> poll , args = ( self . _rdk_producer , self . _stop_poller_thread ) ) <EOL> def stop ( self ) : <EOL> self . _stop_poller_thread . set ( ) <EOL> super ( RdKafkaProducer , self ) . stop ( ) <EOL> self . _rdk_producer . stop ( ) <EOL> def _produce ( self , message ) : <EOL> try : <EOL> self . _rdk_producer . produce ( message ) <EOL> except RdKafkaStoppedException : <EOL> raise ProducerStoppedException <EOL> def _wait_all ( self ) : <EOL> log . info ( "<STR_LIT>" ) <EOL> if self . _poller_thread is not None : <EOL> self . _poller_thread . join ( ) <EOL> def _mk_rdkafka_config_lists ( self ) : <EOL> """<STR_LIT>""" <EOL> map_compression_types = { <EOL> CompressionType . NONE : "<STR_LIT:none>" , <EOL> CompressionType . GZIP : "<STR_LIT>" , <EOL> CompressionType . SNAPPY : "<STR_LIT>" , <EOL> } <EOL> conf = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . _cluster . _socket_timeout_ms , <EOL> "<STR_LIT>" : self . _max_queued_messages , <EOL> "<STR_LIT>" : self . _linger_ms , <EOL> "<STR_LIT>" : self . _max_retries , <EOL> "<STR_LIT>" : self . _retry_backoff_ms , <EOL> "<STR_LIT>" : map_compression_types [ self . _compression ] , <EOL> "<STR_LIT>" : self . _min_queued_messages , <EOL> "<STR_LIT>" : self . _max_request_size , <EOL> "<STR_LIT>" : "<STR_LIT:false>" , <EOL> } <EOL> topic_conf = { <EOL> "<STR_LIT>" : ( <EOL> self . _required_acks if self . _required_acks <= <NUM_LIT:1> else - <NUM_LIT:1> ) , <EOL> "<STR_LIT>" : self . _ack_timeout_ms , <EOL> "<STR_LIT>" : <NUM_LIT> * self . _max_retries * ( <EOL> self . _ack_timeout_ms + self . _retry_backoff_ms ) , <EOL> } <EOL> conf = [ ( key , str ( conf [ key ] ) ) for key in conf ] <EOL> topic_conf = [ ( key , str ( topic_conf [ key ] ) ) for key in topic_conf ] <EOL> return conf , topic_conf </s>
