aboutsummaryrefslogtreecommitdiffstats
path: root/python/fatcat_web/kafka.py
blob: 895e719f9ede3080f7b45b401439d13e08d3d16a (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

import requests

from fatcat_web import Config


def kafka_pixy_produce(topic, msg, key=None, sync=True, timeout=5):
    """
    Simple helper to public a message to the given Kafka topic, via the
    configured kafka-pixy HTTP gateway

    topic: string
    msg: string
    key: optional, bytes
    timeout: seconds
    """

    if not Config.KAFKA_PIXY_ENDPOINT:
        raise Exception("Kafka produce error: kafka-pixy endpoint not configured")

    params = dict()
    if key:
        params['key'] = key
    if sync:
        params['sync'] = True
    resp = requests.post(
        "{}/topics/{}/messages".format(Config.KAFKA_PIXY_ENDPOINT, topic),
        params=params,
        data=msg,
        headers={"Content-Type": "text/plain"},
        timeout=timeout,
    )
    resp.raise_for_status()
    print(resp.json())