Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 34 additions & 28 deletions faststream/redis/publisher/usecase.py
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ async def request(
return msg


class ListPublisher(LogicPublisher):
class _ListPublisherMixin(LogicPublisher):
__slots__ = ("_list",)

def __init__(
Expand All @@ -192,6 +192,36 @@ def subscriber_property(self, *, name_only: bool) -> dict[str, Any]:
"stream": None,
}

@override
async def request(
self,
message: "SendableMessage" = None,
list: str | None = None,
*,
correlation_id: str | None = None,
headers: dict[str, Any] | None = None,
timeout: float | None = 30.0,
) -> "RedisChannelMessage":
cmd = RedisPublishCommand(
message,
list=list or self.list.name,
headers=self.headers | (headers or {}),
correlation_id=correlation_id or self._outer_config.id_generator(),
timeout=timeout,
_publish_type=PublishType.REQUEST,
message_format=self.config.message_format,
)

msg: RedisChannelMessage = await self._basic_request(
cmd,
producer=self.producer,
)
return msg


class ListPublisher(_ListPublisherMixin):
__slots__ = ()

@override
async def publish(
self,
Expand Down Expand Up @@ -242,34 +272,10 @@ async def _publish(
_extra_middlewares=_extra_middlewares,
)

@override
async def request(
self,
message: "SendableMessage" = None,
list: str | None = None,
*,
correlation_id: str | None = None,
headers: dict[str, Any] | None = None,
timeout: float | None = 30.0,
) -> "RedisChannelMessage":
cmd = RedisPublishCommand(
message,
list=list or self.list.name,
headers=self.headers | (headers or {}),
correlation_id=correlation_id or self._outer_config.id_generator(),
timeout=timeout,
_publish_type=PublishType.REQUEST,
message_format=self.config.message_format,
)

msg: RedisChannelMessage = await self._basic_request(
cmd,
producer=self.producer,
)
return msg


class ListBatchPublisher(ListPublisher):
# Not a `ListPublisher`: its `publish(*messages)` would take the `list` that
# `ListPublisher.publish(message, list)` accepts positionally as a second message.
Comment thread
IvanKirpichnikov marked this conversation as resolved.
Resolved
class ListBatchPublisher(_ListPublisherMixin):
__slots__ = ()

@override
Expand Down
18 changes: 17 additions & 1 deletion tests/brokers/redis/test_requests.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import pytest

from faststream import BaseMiddleware
from faststream.redis import BinaryMessageFormatV1
from faststream.redis import BinaryMessageFormatV1, RedisBroker
from tests.brokers.base.requests import RequestsTestcase

from .basic import RedisMemoryTestcaseConfig, RedisTestcaseConfig
Expand Down Expand Up @@ -34,6 +34,22 @@ class RedisRequestsTestcase(RequestsTestcase):
def get_middleware(self, **kwargs: Any) -> Any:
return Mid

async def test_list_publisher_request(self, queue: str) -> None:
broker: RedisBroker = self.get_broker()

publisher = broker.publisher(list=queue)

@broker.subscriber(list=queue)
async def handler(msg: Any) -> str:
return "Response"

async with self.patch_broker(broker):
await broker.start()

response = await publisher.request(None, timeout=self.timeout)

assert await response.decode() == "Response"
Comment thread
IvanKirpichnikov marked this conversation as resolved.
Resolved


@pytest.mark.connected()
@pytest.mark.redis()
Expand Down
9 changes: 9 additions & 0 deletions tests/mypy/redis.py
Comment thread
IvanKirpichnikov marked this conversation as resolved.
Resolved
Original file line number Diff line number Diff line change
Expand Up @@ -536,3 +536,12 @@ def accepts_any_list_sub(list_sub: ListSub) -> None: ...


accepts_any_list_sub(ListSub("test", batch=True))


def accepts_list_publisher(publisher: ListPublisher) -> None: ...


# `publish(*messages)` of a batch publisher would take the `list` passed
# positionally to `ListPublisher.publish` as a second message
batch_publisher = RedisBroker().publisher(list=ListSub("test", batch=True))
accepts_list_publisher(batch_publisher) # type: ignore[arg-type]
Loading