Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix latest EventBroadcaster change to keep original behavior #78

Merged
merged 2 commits into from
Jul 2, 2024
Merged

Fix latest EventBroadcaster change to keep original behavior #78

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
212aafa
Fix EventBroadcaster change to keep original behavior
roekatz Jun 27, 2024
1d06fd3
Bump version to 0.3.9
roekatz Jun 27, 2024
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
72 changes: 37 additions & 35 deletions fastapi_websocket_pubsub/event_broadcaster.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -222,9 +222,6 @@ async def __aenter__(self):
return await self._context_manager.__aenter__()

async def __aexit__(self, exc_type, exc, tb):
if self.listening_broadcast_channel is not None:
await self.listening_broadcast_channel.disconnect()
self.listening_broadcast_channel = None
await self._context_manager.__aexit__(exc_type, exc, tb)

async def start_reader_task(self):
Expand Down Expand Up @@ -268,40 +265,45 @@ async def __read_notifications__(self):
read incoming broadcasts and posting them to the intreal notifier
"""
logger.debug("Starting broadcaster listener")
# Subscribe to our channel
async with self.listening_broadcast_channel.subscribe(
channel=self._channel
) as subscriber:
async for event in subscriber:
try:
notification = BroadcastNotification.parse_raw(event.message)
# Avoid re-publishing our own broadcasts
if notification.notifier_id != self._id:
logger.debug(
"Handling incoming broadcast event: {}".format(
{
"topics": notification.topics,
"src": notification.notifier_id,
}
try:
# Subscribe to our channel
async with self.listening_broadcast_channel.subscribe(
channel=self._channel
) as subscriber:
async for event in subscriber:
try:
notification = BroadcastNotification.parse_raw(event.message)
# Avoid re-publishing our own broadcasts
if notification.notifier_id != self._id:
logger.debug(
"Handling incoming broadcast event: {}".format(
{
"topics": notification.topics,
"src": notification.notifier_id,
}
)
)
)
# Notify subscribers of message received from broadcast
task = asyncio.create_task(
self._notifier.notify(
notification.topics,
notification.data,
notifier_id=self._id,
# Notify subscribers of message received from broadcast
task = asyncio.create_task(
self._notifier.notify(
notification.topics,
notification.data,
notifier_id=self._id,
)
)
)

self._tasks.add(task)
self._tasks.add(task)

def cleanup(task):
self._tasks.remove(task)
def cleanup(task):
self._tasks.remove(task)

task.add_done_callback(cleanup)
except:
logger.exception("Failed handling incoming broadcast")
logger.info(
"No more events to read from subscriber (underlying connection closed)"
)
task.add_done_callback(cleanup)
except:
logger.exception("Failed handling incoming broadcast")
logger.info(
"No more events to read from subscriber (underlying connection closed)"
)
finally:
if self.listening_broadcast_channel is not None:
await self.listening_broadcast_channel.disconnect()
self.listening_broadcast_channel = None
2 changes: 1 addition & 1 deletion setup.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ def get_requirements(env=""):

setup(
name="fastapi_websocket_pubsub",
version="0.3.8",
version="0.3.9",
author="Or Weis",
author_email="[email protected]",
description="A fast and durable PubSub channel over Websockets (using fastapi-websockets-rpc).",
Expand Down
Loading