Skip to content

Conversation

@tacaswell
Copy link
Member

This is a mass-up of caproto-monitor and the guts of KafkaPublisher.

To read the values back out:

from functools import partial
from confluent_kafka import Consumer
import msgpack
import msgpack_numpy as mpn


consumer_config = {
    "bootstrap.servers": "127.0.0.1:9092",
    "group.id": "kafka-unit-test-group-id",
    "auto.offset.reset": "latest",
}


deserializer = partial(msgpack.loads, object_hook=mpn.decode)

consumer = Consumer(consumer_config)
consumer.subscribe(["demo"])

while True:
    msg = consumer.poll(1)

    if msg is None:
        # no message was delivered
        # do some work before polling again
        time.sleep(0.1)
    else:
        payload = deserializer(msg.value())
        print(payload)

Not really sure where to send this (bluesky-pods? caproto?), settled on here but happy to move.

@codecov-commenter
Copy link

codecov-commenter commented Aug 28, 2020

Codecov Report

Merging #103 into master will increase coverage by 0.56%.
The diff coverage is n/a.

Impacted file tree graph

@@            Coverage Diff             @@
##           master     #103      +/-   ##
==========================================
+ Coverage   56.61%   57.17%   +0.56%     
==========================================
  Files          13       13              
  Lines         892      892              
==========================================
+ Hits          505      510       +5     
+ Misses        387      382       -5     
Impacted Files Coverage Δ
nslsii/_version.py 44.80% <0.00%> (+1.79%) ⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 18c0632...8122ee7. Read the comment docs.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants