faust.channels

Channel.

A channel is used to send values to streams.

The stream will iterate over incoming events in the channel.

class faust.channels.Channel(app: faust.types.app.AppT, *, key_type: Union[typing.Type[faust.types.models.ModelT], typing.Type[bytes], typing.Type[str]] = None, value_type: Union[typing.Type[faust.types.models.ModelT], typing.Type[bytes], typing.Type[str]] = None, is_iterator: bool = False, queue: mode.utils.queues.ThrowableQueue = None, maxsize: int = None, root: faust.types.channels.ChannelT = None, active_partitions: Set[faust.types.tuples.TP] = None, loop: asyncio.events.AbstractEventLoop = None) → None

Create new channel.

Parameters:

• app (AppT[]) – The app that created this channel (app.channel())
• key_type (Union[Type[ModelT], Type[bytes], Type[str], None]) – The Model used for keys in this channel.
• value_type (Union[Type[ModelT], Type[bytes], Type[str], None]) – The Model used for values in this channel.
• maxsize (Optional[int]) – The maximum number of messages this channel can hold. If exceeded any new put call will block until a message is removed from the channel.
• loop (Optional[AbstractEventLoop]) – The asyncio event loop to use.

coroutine deliver(self, message: faust.types.tuples.Message) → None

Return type:None

queue

clone(*, is_iterator: bool = None, **kwargs) → faust.types.channels.ChannelT

Return type:ChannelT[]

stream(**kwargs) → faust.types.streams.StreamT

Create stream reading from this channel.

Return type:StreamT[+T_co]

get_topic_name() → str

Return type:str

as_future_message(key: Union[bytes, faust.types.core.ModelT, typing.Any, NoneType] = None, value: Union[bytes, faust.types.core.ModelT, typing.Any] = None, partition: int = None, key_serializer: Union[faust.types.codecs.CodecT, str, NoneType] = None, value_serializer: Union[faust.types.codecs.CodecT, str, NoneType] = None, callback: Callable[faust.types.tuples.FutureMessage, Union[NoneType, typing.Awaitable[NoneType]]] = None) → faust.types.tuples.FutureMessage

Return type:FutureMessage[]

prepare_key(key: Union[bytes, faust.types.core.ModelT, typing.Any, NoneType], key_serializer: Union[faust.types.codecs.CodecT, str, NoneType]) → Any

Return type:Any

prepare_value(value: Union[bytes, faust.types.core.ModelT, typing.Any], value_serializer: Union[faust.types.codecs.CodecT, str, NoneType]) → Any

Return type:Any

empty() → bool

Return type:bool

coroutine declare(self) → None

Return type:None

coroutine decode(self, message: faust.types.tuples.Message, *, propagate: bool = False) → faust.types.events.EventT

Return type:EventT[]

coroutine get(self, *, timeout: Union[datetime.timedelta, float, str] = None) → Any

Return type:Any

maybe_declare

coroutine on_decode_error(self, exc: Exception, message: faust.types.tuples.Message) → None

Return type:None

coroutine on_key_decode_error(self, exc: Exception, message: faust.types.tuples.Message) → None

Return type:None

on_stop_iteration() → None

Return type:None

coroutine on_value_decode_error(self, exc: Exception, message: faust.types.tuples.Message) → None

Return type:None

coroutine publish_message(self, fut: faust.types.tuples.FutureMessage, wait: bool = True) → Awaitable[faust.types.tuples.RecordMetadata]

Return type:Awaitable[RecordMetadata]

coroutine put(self, value: Any) → None

Return type:None

coroutine send(self, key: Union[bytes, faust.types.core.ModelT, typing.Any, NoneType] = None, value: Union[bytes, faust.types.core.ModelT, typing.Any] = None, partition: int = None, key_serializer: Union[faust.types.codecs.CodecT, str, NoneType] = None, value_serializer: Union[faust.types.codecs.CodecT, str, NoneType] = None, callback: Callable[faust.types.tuples.FutureMessage, Union[NoneType, typing.Awaitable[NoneType]]] = None, force: bool = False) → Awaitable[faust.types.tuples.RecordMetadata]

Send message to channel.

Return type:Awaitable[RecordMetadata]

coroutine throw(self, exc: BaseException) → None

Return type:None

derive(**kwargs) → faust.types.channels.ChannelT

Return type:ChannelT[]

subscriber_count

label