Source code for platypush.plugins.kafka

import json
import logging

from platypush.context import get_backend
from platypush.plugins import Plugin, action


[docs]class KafkaPlugin(Plugin): """ Plugin to send messages to an Apache Kafka instance (https://kafka.apache.org/) Triggers: * :class:`platypush.message.event.kafka.KafkaMessageEvent` when a new message is received on the consumer topic. Requires: * **kafka** (``pip install kafka-python``) """
[docs] def __init__(self, server=None, port=9092, **kwargs): """ :param server: Default Kafka server name or address. If None (default), then it has to be specified upon message sending. :type server: str :param port: Default Kafka server port (default: 9092). :type port: int """ super().__init__(**kwargs) self.server = '{server}:{port}'.format(server=server, port=port) \ if server else None self.producer = None # Kafka can be veryyyy noisy logging.getLogger('kafka').setLevel(logging.ERROR)
[docs] @action def send_message(self, msg, topic, server=None): """ :param msg: Message to send - as a string, bytes stream, JSON, Platypush message, dictionary, or anything that implements ``__str__`` :param topic: Topic to send the message to. :type topic: str :param server: Kafka server name or address + port (format: ``host:port``). If None, then the default server will be used :type server: str """ from kafka import KafkaProducer if not server: if not self.server: try: kafka_backend = get_backend('kafka') server = kafka_backend.server except Exception as e: raise RuntimeError(f'No Kafka server nor default server specified: {str(e)}') else: server = self.server if isinstance(msg, dict) or isinstance(msg, list): msg = json.dumps(msg) msg = str(msg).encode('utf-8') producer = KafkaProducer(bootstrap_servers=server) producer.send(topic, msg) producer.flush()
# vim:sw=4:ts=4:et: