Kafka usage examples
Kafka topics - consuming messages through channels and publishing with self.out.kafka.
Consuming messages from topics
No programming is needed to consume messages from Kafka topics. Create a Kafka channel in Dashboard, point it to a topic, and a service of your choice will be invoked for each message received.
The message payload will be available in self.request.input.
from zato.server.service import Service
class MyService(Service):
def handle(self):
# The incoming Kafka message payload
data = self.request.input
# Log what was received
self.logger.info('Kafka message: %s', data)
Publishing messages to topics
Create a Kafka outgoing connection in Dashboard and use self.out.kafka to publish messages.
The connection is looked up by name, and .send() accepts strings, bytes, dicts, lists or any other JSON-serializable object.
from zato.server.service import Service
class MyService(Service):
def handle(self):
# Get a connection by name
conn = self.out.kafka['my-publisher']
# Send a string
conn.send('Hello from Zato')
Sending structured data
When you pass a dict or list, it is automatically serialized to JSON before publishing.
from zato.server.service import Service
class MyService(Service):
def handle(self):
conn = self.out.kafka['my-publisher']
conn.send({
'event': 'order.created',
'order_id': '12345',
'customer': 'Acme Corp',
})
Processing and forwarding
A common pattern is to consume a message from one topic, process it and publish the result to another.
from zato.server.service import Service
class MyService(Service):
def handle(self):
# Incoming message from a Kafka channel
data = self.request.input
self.logger.info('Processing: %s', data)
# Forward to another topic through a different outgoing connection
conn = self.out.kafka['processed-events']
conn.send(data)