使用 faust 流访问 grpc 回复中的 kafka 标头答案

【问题标题】：Access kafka headers in rpc replies using faust streamming使用 faust 流访问 grpc 回复中的 kafka 标头
【发布时间】：2020-03-25 14:25:18
【问题描述】：

是否可以在 faust rpc 回复中访问 kafka 标头？这是两个浮士德代理的示例。一个（pow）调用另一个（mul）并接收一个结果作为值。但是怎么知道reply topic里面的kafka headers呢？

#!/usr/bin/env python
from typing import AsyncIterable
import faust
from faust import StreamT

app = faust.App('RPC99', reply_create_topic=True)
pow_topic = app.topic('RPC__pow')
mul_topic = app.topic('RPC__mul')

@app.agent(pow_topic)
async def pow(stream: StreamT[float]) -> AsyncIterable[float]:
    async for value in stream:
        yield await mul.ask(value=value ** 2)
        # Headers for the returning result here?

@app.agent(mul_topic)
async def mul(stream: StreamT[float]) -> AsyncIterable[float]:
    async for value in stream:
        yield value * 100.0

【问题讨论】：

标签： faust

【解决方案1】：

您可以从收到的事件中获取标头：

stream.events()

@app.agent(topic)
async def iterrate(stream):
    async for event in stream.events():
        value = event.value
        offset = event.message.offset
        headers = event.headers

faust.types.events 的定义：

https://faust.readthedocs.io/en/latest/reference/faust.types.events.html#faust.types.events.EventT

【讨论】：

【解决方案2】：

流事件似乎具有headers 类型的Union[List[Tuple[str, bytes]] 属性。

docs 在这里有它，但目前它们并没有非常详细地说明这一点。

看起来你可以做类似的事情

@app.agent(topic)
async def process(stream):
    async for value in stream:
        do_something(value.headers)

如果您已经反序列化了消息，看起来您还可以通过faust.streams.current_event() 使用streams api 访问原始事件。

根据Роман 的评论进行编辑：

看起来您必须显式发送消息才能修改标头。在channel api 中查看send() 的函数签名。我认为 send 或该类中 send 的其他变体之一应该是您正在寻找的。p>

【讨论】：

是的，我知道我可以从 stream.events() 或使用 current_event() 访问它们。但是这些状态用于传入消息。我问的是回复。
ask() 方法也有headers