Appending streams during runtime to stream consumers

[rahulswaminathan]

Hello @alisaifee,

In stream consumer API (both standalone and group), all streams and its parameters have to be sent during Consumer/GroupConsumer initialisation. I was thinking if there could be a public method to add streams after initialisation too like add_stream. What do you think?

[alisaifee]

Hi @rahulswaminathan

It should be possible, basically we'd have to allow a kind of re-initialization when a new stream is added to the consumer. Curious though, what kind of use-case would this be beneficial for?

[rahulswaminathan]

I'm working on a stream aggregator where the complete list of streams might not be available during initialisation and thus, the aggregator should be able to consume a new stream during runtime.

[alisaifee]

@rahulswaminathan I think it would look something like this

[rahulswaminathan]

Awesome! Just one more thing, could we add another parameter to accept the stream parameters, just like in init method?

async def add_stream(self, stream: StringT, **stream_parameters: StreamParameters) -> None:
"""
Adds a new stream identifier to this consumer
:param stream: The stream identifier
:param stream_parameters: Mapping of optional parameters to use by stream in :paramref: stream.
"""
self.streams.add(stream)
self.state.update({stream: stream_parameters.get(nativestr(stream), {})})
await self.initialize(partial=True)

[alisaifee]

I went with just adding identifier: Optional[StringT] to keep it simple for now. Keep in mind that providing the identifier when using the GroupConsumer means something entirely different and for that reason the default for GroupConsumer.add_stream will be explicitly set to >.

[alisaifee]

@rahulswaminathan this change is now in master and will be available in the upcoming 4.12.0b4 release if you would like to give it a try. The associated documentation is here

[rahulswaminathan]

Thanks @alisaifee! I'll try it out.