1414# License for the specific language governing permissions and limitations
1515# under the License.
1616
17- """Module containing a Logstask cASO messenger."""
17+ """Module containing a Kafka cASO messenger."""
1818
1919import socket
2020
@@ -49,7 +49,7 @@ class KafkaMessenger(caso.messenger.BaseMessenger):
4949 """Format and send records to a kafka host."""
5050
5151 def __init__ (self , brokers = CONF .kafka .brokers , topic = CONF .kafka .topic , serviceName = CONF .kafka .serviceName , username = CONF .kafka .username , password = CONF .kafka .password ):
52- """Get a logstash messenger for a given host and port."""
52+ """Get a kafka messenger for a given host and port."""
5353 super (KafkaMessenger , self ).__init__ ()
5454 self .brokers = CONF .kafka .brokers
5555 self .topic = CONF .kafka .topic
@@ -71,7 +71,7 @@ def delivery_report(self, err, msg):
7171 def push (self , records ):
7272
7373 # NOTE(acostantini): code for the serialization and push of the
74- # records in logstash. JSON format to be used and encoding UTF-8
74+ # records in logstash via kafka . JSON format to be used and encoding UTF-8
7575 """Serialization of records to be sent to logstash via kafka"""
7676 if not records :
7777 return
@@ -90,11 +90,11 @@ def push(self, records):
9090 'sasl.username' : self .username ,
9191 'sasl.password' : self .password
9292
93- # We can tune as args batch_size and linger_ms
93+ # Producer
9494 producer = Producer (** conf )
9595
9696
97- """Push records to logstash using tcp ."""
97+ """Push records to be serialized using logstash_message definition ."""
9898 for record in records :
9999 #serialization of record
100100 rec = record .logstash_message ()
0 commit comments