Produce event interface

[woile]

Goal

Define the interface to produce events

Considerations

• should it be possible to emit to a different broker (let's say we consume from kafka but we want to emit to a redis stream, what then?)

Ideas

Names

emit
produce
send
publish

1. Dependency injection

async def hello(stream: Stream, emit: Callable):
 async for value in stream:
 emit("a-topic", value)

2. Import function

from dam import emit

async def hello(stream: Stream):
 async for value in stream:
 emit("a-topic", value)

Decisions

• keyword: send
• interface: TBD

[woile]

@marcosschroh suggestions?

[marcosschroh]

I think we can go for option 1, and use send as a name as most of python streaming libraries has a send function.

Maybe something like this:

class Producer:
 ....

 async def send(self, topic, value, key=None, partition=None, timestamp_ms=None, headers=None):
 ...


# When the topic is created and is the signature has the send argument, we inject it.
async def hello(stream: Stream, send: Awaitable):
 async for value in stream:
 await send("a-topic", value)

What do you think?

[marcosschroh]

Also, do we need a singleton for the producer ? I am seeing only one producer instance to be shared

[woile]

I'm okay with send as keyword.

Why do you think a singleton is needed? The kafka producer will have to be instanciated every time probably, or at least per topic, what would we gain frrom a singleton?

I think we should also support this behavior:

app = App(broker="kafka://kafka:29000")
app.send("topic", b"payload")

For simplicity and testing purposes