Enable presence again. Fix up api to match old api.

paul/schema_breaking_changes
Erik Johnston 2014-08-26 19:40:29 +01:00
parent c1cf0b334e
commit 67c5f89244
5 changed files with 35 additions and 10 deletions

View File

@ -15,6 +15,8 @@
from twisted.internet import defer from twisted.internet import defer
from synapse.api.events import SynapseEvent
from ._base import BaseHandler from ._base import BaseHandler
import logging import logging
@ -50,10 +52,15 @@ class EventStreamHandler(BaseHandler):
events, tokens = yield self.notifier.get_events_for(auth_user, pagin_config, timeout) events, tokens = yield self.notifier.get_events_for(auth_user, pagin_config, timeout)
chunks = [
e.get_dict() if isinstance(e, SynapseEvent) else e
for e in events
]
chunk = { chunk = {
"chunk": [e.get_dict() for e in events], "chunk": chunks,
"start_token": tokens[0].to_string(), "start": tokens[0].to_string(),
"end_token": tokens[1].to_string(), "end": tokens[1].to_string(),
} }
defer.returnValue(chunk) defer.returnValue(chunk)

View File

@ -142,7 +142,7 @@ class PresenceHandler(BaseHandler):
@defer.inlineCallbacks @defer.inlineCallbacks
def is_presence_visible(self, observer_user, observed_user): def is_presence_visible(self, observer_user, observed_user):
defer.returnValue(True) defer.returnValue(True)
return #return
# FIXME (erikj): This code path absolutely kills the database. # FIXME (erikj): This code path absolutely kills the database.
assert(observed_user.is_mine) assert(observed_user.is_mine)
@ -189,7 +189,7 @@ class PresenceHandler(BaseHandler):
@defer.inlineCallbacks @defer.inlineCallbacks
def set_state(self, target_user, auth_user, state): def set_state(self, target_user, auth_user, state):
return # return
# TODO (erikj): Turn this back on. Why did we end up sending EDUs # TODO (erikj): Turn this back on. Why did we end up sending EDUs
# everywhere? # everywhere?

View File

@ -126,8 +126,8 @@ class MessageHandler(BaseHandler):
chunk = { chunk = {
"chunk": [e.get_dict() for e in events], "chunk": [e.get_dict() for e in events],
"start_token": from_token.to_string(), "start": from_token.to_string(),
"end_token": next_token.to_string(), "end": next_token.to_string(),
} }
defer.returnValue(chunk) defer.returnValue(chunk)

View File

@ -91,7 +91,25 @@ class Notifier(object):
) )
def on_new_user_event(self, *args, **kwargs): def on_new_user_event(self, *args, **kwargs):
pass source = self.event_sources.sources[1]
listeners = self.signal_keys_to_users.get(
(source.SIGNAL_NAME, "moose"),
[]
)
for listener in listeners:
events, end_token = yield source.get_new_events_for_user(
listener.user,
listener.from_token,
listener.limit,
key="moose",
)
if events:
listener.notify(
self, events, listener.from_token, end_token
)
def get_events_for(self, user, pagination_config, timeout): def get_events_for(self, user, pagination_config, timeout):
deferred = defer.Deferred() deferred = defer.Deferred()

View File

@ -97,7 +97,7 @@ class PresenceStreamSource(object):
data = [x[1].make_event(user=x[0], clock=clock) for x in updates] data = [x[1].make_event(user=x[0], clock=clock) for x in updates]
end_token = from_token.copy_and_replace( end_token = from_token.copy_and_replace(
"presence_key", latest_serial "presence_key", latest_serial + 1
) )
return ((data, end_token)) return ((data, end_token))
else: else:
@ -107,7 +107,7 @@ class PresenceStreamSource(object):
return (([], end_token)) return (([], end_token))
def get_keys_for_user(self, user): def get_keys_for_user(self, user):
return defer.succeed([]) return defer.succeed(["moose"])
def get_current_token_part(self): def get_current_token_part(self):
presence = self.hs.get_handlers().presence_handler presence = self.hs.get_handlers().presence_handler