faust.transport.drivers.memory

Experimental: In-memory transport.

class faust.transport.drivers.memory.RebalanceListener(*args, **kwargs)[source]

In-memory rebalance listener.

class faust.transport.drivers.memory.Consumer(transport: faust.types.transports.TransportT, callback: Callable[faust.types.tuples.Message, Awaitable], on_partitions_revoked: Callable[Set[faust.types.tuples.TP], Awaitable[None]], on_partitions_assigned: Callable[Set[faust.types.tuples.TP], Awaitable[None]], *, commit_interval: float = None, commit_livelock_soft_timeout: float = None, loop: asyncio.events.AbstractEventLoop = None, **kwargs) → None[source]

In-memory consumer.

class RebalanceListener(*args, **kwargs)

In-memory rebalance listener.

consumer_stopped_errors = ()
pause_partitions(tps: Iterable[faust.types.tuples.TP]) → None[source]
Return type:None
resume_partitions(partitions: Iterable[faust.types.tuples.TP]) → None[source]
Return type:None
assignment() → Set[faust.types.tuples.TP][source]
Return type:Set[TP]
highwater(tp: faust.types.tuples.TP) → int[source]
Return type:int
coroutine create_topic(self, topic: str, partitions: int, replication: int, *, config: Mapping[str, Any] = None, timeout: Union[datetime.timedelta, float, str] = 1000.0, retention: Union[datetime.timedelta, float, str] = None, compacting: bool = None, deleting: bool = None, ensure_created: bool = False) → None[source]
Return type:None
coroutine earliest_offsets(self, *partitions) → MutableMapping[faust.types.tuples.TP, int][source]
Return type:MutableMapping[TP, int]
getmany(timeout: float) → AsyncIterator[Tuple[faust.types.tuples.TP, faust.types.tuples.Message]][source]
Return type:AsyncIterator[Tuple[TP, Message]]
coroutine highwaters(self, *partitions) → MutableMapping[faust.types.tuples.TP, int][source]
Return type:MutableMapping[TP, int]
logger = <Logger faust.transport.drivers.memory (WARNING)>
coroutine perform_seek(self) → None[source]
Return type:None
coroutine position(self, tp: faust.types.tuples.TP) → Optional[int][source]
Return type:Optional[int]
coroutine seek(self, partition: faust.types.tuples.TP, offset: int) → None[source]
Return type:None
coroutine seek_to_beginning(self, *partitions) → None[source]
Return type:None
coroutine seek_to_latest(self, *partitions) → None[source]
Return type:None
coroutine subscribe(self, topics: Iterable[str]) → None[source]
Return type:None
class faust.transport.drivers.memory.Producer(transport: faust.types.transports.TransportT, loop: asyncio.events.AbstractEventLoop = None, **kwargs) → None[source]

In-memory producer.

coroutine create_topic(self, topic: str, partitions: int, replication: int, *, config: Mapping[str, Any] = None, timeout: Union[datetime.timedelta, float, str] = None, retention: Union[datetime.timedelta, float, str] = None, compacting: bool = None, deleting: bool = None, ensure_created: bool = False) → None[source]
Return type:None
logger = <Logger faust.transport.drivers.memory (WARNING)>
coroutine send(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int], timestamp: Optional[float], *, transactional_id: str = None) → Awaitable[faust.types.tuples.RecordMetadata][source]
Return type:Awaitable[RecordMetadata]
coroutine send_and_wait(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int], timestamp: Optional[float], *, transactional_id: str = None) → faust.types.tuples.RecordMetadata[source]
Return type:RecordMetadata
class faust.transport.drivers.memory.Transport(*args, **kwargs) → None[source]

In-memory transport.

class Consumer(transport: faust.types.transports.TransportT, callback: Callable[faust.types.tuples.Message, Awaitable], on_partitions_revoked: Callable[Set[faust.types.tuples.TP], Awaitable[None]], on_partitions_assigned: Callable[Set[faust.types.tuples.TP], Awaitable[None]], *, commit_interval: float = None, commit_livelock_soft_timeout: float = None, loop: asyncio.events.AbstractEventLoop = None, **kwargs) → None

In-memory consumer.

class RebalanceListener(*args, **kwargs)

In-memory rebalance listener.

assignment() → Set[faust.types.tuples.TP]
Return type:Set[TP]
consumer_stopped_errors = ()
coroutine create_topic(self, topic: str, partitions: int, replication: int, *, config: Mapping[str, Any] = None, timeout: Union[datetime.timedelta, float, str] = 1000.0, retention: Union[datetime.timedelta, float, str] = None, compacting: bool = None, deleting: bool = None, ensure_created: bool = False) → None
Return type:None
coroutine earliest_offsets(self, *partitions) → MutableMapping[faust.types.tuples.TP, int]
Return type:MutableMapping[TP, int]
getmany(timeout: float) → AsyncIterator[Tuple[faust.types.tuples.TP, faust.types.tuples.Message]]
Return type:AsyncIterator[Tuple[TP, Message]]
highwater(tp: faust.types.tuples.TP) → int
Return type:int
coroutine highwaters(self, *partitions) → MutableMapping[faust.types.tuples.TP, int]
Return type:MutableMapping[TP, int]
logger = <Logger faust.transport.drivers.memory (WARNING)>
pause_partitions(tps: Iterable[faust.types.tuples.TP]) → None
Return type:None
coroutine perform_seek(self) → None
Return type:None
coroutine position(self, tp: faust.types.tuples.TP) → Optional[int]
Return type:Optional[int]
resume_partitions(partitions: Iterable[faust.types.tuples.TP]) → None
Return type:None
coroutine seek(self, partition: faust.types.tuples.TP, offset: int) → None
Return type:None
coroutine seek_to_beginning(self, *partitions) → None
Return type:None
coroutine seek_to_latest(self, *partitions) → None
Return type:None
coroutine subscribe(self, topics: Iterable[str]) → None
Return type:None
class Producer(transport: faust.types.transports.TransportT, loop: asyncio.events.AbstractEventLoop = None, **kwargs) → None

In-memory producer.

coroutine create_topic(self, topic: str, partitions: int, replication: int, *, config: Mapping[str, Any] = None, timeout: Union[datetime.timedelta, float, str] = None, retention: Union[datetime.timedelta, float, str] = None, compacting: bool = None, deleting: bool = None, ensure_created: bool = False) → None
Return type:None
logger = <Logger faust.transport.drivers.memory (WARNING)>
coroutine send(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int], timestamp: Optional[float], *, transactional_id: str = None) → Awaitable[faust.types.tuples.RecordMetadata]
Return type:Awaitable[RecordMetadata]
coroutine send_and_wait(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int], timestamp: Optional[float], *, transactional_id: str = None) → faust.types.tuples.RecordMetadata
Return type:RecordMetadata
default_port = 9092
driver_version = 'memory-1.5.0b1'
coroutine send(self, topic: str, key: Optional[bytes], value: Optional[bytes], partition: Optional[int], timestamp: Optional[float]) → faust.types.tuples.RecordMetadata[source]
Return type:RecordMetadata
coroutine subscribe(self, topics: Iterable[str]) → None[source]
Return type:None