Skip to content

proxystore.stream.shims.kafka

proxystore/stream/shims/kafka.py

Kafka publisher and subscriber shims.

Shims to the kafka-python package.

KafkaPublisher 

KafkaPublisher(client: KafkaProducer)

Kafka publisher shim.

Parameters:

Source code in proxystore/stream/shims/kafka.py 
26
27
def __init__(self, client: kafka.KafkaProducer) -> None:
    self.client = client

close() 

close() -> None

Close this publisher.

Source code in proxystore/stream/shims/kafka.py 
29
30
31
def close(self) -> None:
    """Close this publisher."""
    self.client.close()

send() 

send(topic: str, message: bytes) -> None

Publish a message to the stream.

Parameters:

  • topic (str) –

    Stream topic to publish message to.

  • message (bytes) –

    Message as bytes to publish to the stream.

Source code in proxystore/stream/shims/kafka.py 
33
34
35
36
37
38
39
40
41
def send(self, topic: str, message: bytes) -> None:
    """Publish a message to the stream.

    Args:
        topic: Stream topic to publish message to.
        message: Message as bytes to publish to the stream.
    """
    future = self.client.send(topic, message)
    future.get()

KafkaSubscriber 

KafkaSubscriber(client: KafkaConsumer)

Kafka subscriber shim.

This shim is an iterable object which will yield bytes messages from the stream, blocking on the next message, until the stream is closed.

Parameters:

Source code in proxystore/stream/shims/kafka.py 
55
56
def __init__(self, client: kafka.KafkaConsumer) -> None:
    self.client = client

close() 

close() -> None

Close this subscriber.

Source code in proxystore/stream/shims/kafka.py 
65
66
67
def close(self) -> None:
    """Close this subscriber."""
    self.client.close()