litecord.pubsub: add FriendDispatcher

- gateway.websocket: subscribe to friends on startup
 - presence: dispatch to friends on dispatch_pres
 - storage: add Storage.get_friend_ids
This commit is contained in:
Luna Mendes 2018-10-12 16:17:29 -03:00
parent 46fac95979
commit d28c0f1bc6
6 changed files with 67 additions and 13 deletions

View File

@ -4,7 +4,7 @@ from typing import Any
from logbook import Logger from logbook import Logger
from .pubsub import GuildDispatcher, MemberDispatcher, \ from .pubsub import GuildDispatcher, MemberDispatcher, \
UserDispatcher, ChannelDispatcher UserDispatcher, ChannelDispatcher, FriendDispatcher
log = Logger(__name__) log = Logger(__name__)
@ -20,8 +20,7 @@ class EventDispatcher:
'member': MemberDispatcher(self), 'member': MemberDispatcher(self),
'channel': ChannelDispatcher(self), 'channel': ChannelDispatcher(self),
'user': UserDispatcher(self), 'user': UserDispatcher(self),
'friend': FriendDispatcher(self),
# TODO: channel, friends
} }
async def action(self, backend_str: str, action: str, key, identifier): async def action(self, backend_str: str, action: str, key, identifier):

View File

@ -285,8 +285,8 @@ class GatewayWebsocket:
self.state.user_id self.state.user_id
) )
async def subscribe_guilds(self): async def subscribe_all(self):
"""Subscribe to all available guilds and DM channels. """Subscribe to all guilds, DM channels, and friends.
Subscribing to channels is already handled Subscribing to channels is already handled
by GuildDispatcher.sub by GuildDispatcher.sub
@ -304,6 +304,13 @@ class GatewayWebsocket:
log.info('subscribing to {} dms', len(dm_ids)) log.info('subscribing to {} dms', len(dm_ids))
await self.ext.dispatcher.sub_many('channel', user_id, dm_ids) await self.ext.dispatcher.sub_many('channel', user_id, dm_ids)
# subscribe to all friends
# (their friends will also subscribe back
# when they come online)
friend_ids = await self.storage.get_friend_ids(user_id)
log.info('subscribing to {} friends', len(friend_ids))
await self.ext.dispatcher.sub_many('friend', user_id, friend_ids)
async def update_status(self, status: dict): async def update_status(self, status: dict):
"""Update the status of the current websocket connection.""" """Update the status of the current websocket connection."""
if status is None: if status is None:
@ -398,7 +405,7 @@ class GatewayWebsocket:
self.ext.state_manager.insert(self.state) self.ext.state_manager.insert(self.state)
await self.update_status(presence) await self.update_status(presence)
await self.subscribe_guilds() await self.subscribe_all()
await self.dispatch_ready() await self.dispatch_ready()
async def handle_3(self, payload: Dict[str, Any]): async def handle_3(self, payload: Dict[str, Any]):

View File

@ -96,9 +96,6 @@ class PresenceManager:
"""Dispatch a Presence update to an entire guild.""" """Dispatch a Presence update to an entire guild."""
state = dict(new_state) state = dict(new_state)
if state['status'] == 'invisible':
state['status'] = 'offline'
member = await self.storage.get_member_data_one(guild_id, user_id) member = await self.storage.get_member_data_one(guild_id, user_id)
game = state['game'] game = state['game']
@ -117,14 +114,33 @@ class PresenceManager:
} }
) )
async def dispatch_pres(self, user_id: int, state): async def dispatch_pres(self, user_id: int, state: dict):
"""Dispatch a new presence to all guilds the user is in.""" """Dispatch a new presence to all guilds the user is in.
# TODO: account for sharding
Also dispatches the presence to all the users' friends
"""
if state['status'] == 'invisible':
state['status'] = 'offline'
guild_ids = await self.storage.get_user_guilds(user_id) guild_ids = await self.storage.get_user_guilds(user_id)
for guild_id in guild_ids: for guild_id in guild_ids:
await self.dispatch_guild_pres(guild_id, user_id, state) await self.dispatch_guild_pres(guild_id, user_id, state)
# dispatch to all friends that are subscribed to them
user = await self.storage.get_user(user_id)
game = state['game']
await self.dispatcher.dispatch(
'friend', user_id, 'PRESENCE_UPDATE', {
'user': user,
'status': state['status'],
# rich presence stuff
'game': game,
'activities': [game] if game else []
})
async def friend_presences(self, friend_ids: int) -> List[Dict[str, Any]]: async def friend_presences(self, friend_ids: int) -> List[Dict[str, Any]]:
"""Fetch presences for a group of users. """Fetch presences for a group of users.

View File

@ -2,6 +2,8 @@ from .guild import GuildDispatcher
from .member import MemberDispatcher from .member import MemberDispatcher
from .user import UserDispatcher from .user import UserDispatcher
from .channel import ChannelDispatcher from .channel import ChannelDispatcher
from .friend import FriendDispatcher
__all__ = ['GuildDispatcher', 'MemberDispatcher', __all__ = ['GuildDispatcher', 'MemberDispatcher',
'UserDispatcher', 'ChannelDispatcher'] 'UserDispatcher', 'ChannelDispatcher',
'FriendDispatcher']

22
litecord/pubsub/friend.py Normal file
View File

@ -0,0 +1,22 @@
from logbook import Logger
from .dispatcher import DispatcherWithState
log = Logger(__name__)
class FriendDispatcher(DispatcherWithState):
KEY_TYPE = int
VAL_TYPE = int
async def dispatch(self, user_id: int, event, data):
"""Dispatch an event to all of a users' friends."""
peer_ids = self.state[user_id]
dispatched = 0
for peer_id in peer_ids:
dispatched += await self.main_dispatcher.dispatch(
'user', peer_id, event, data)
log.info('dispatched uid={} {!r} to {} states',
user_id, event, dispatched)

View File

@ -771,6 +771,14 @@ class Storage:
return res return res
async def get_friend_ids(self, user_id: int) -> List[int]:
"""Get all friend IDs for a user."""
rels = await self.get_relationships(user_id)
return [int(r['user']['id'])
for r in rels
if r['type'] == RelationshipType.FRIEND.value]
async def get_dm(self, dm_id: int, user_id: int = None): async def get_dm(self, dm_id: int, user_id: int = None):
dm_chan = await self.get_channel(dm_id) dm_chan = await self.get_channel(dm_id)