Don't make the pushers' event streams cause people to appear online
parent
2cb30767fa
commit
e3e2fc3255
|
@ -47,11 +47,11 @@ class EventStreamHandler(BaseHandler):
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
@log_function
|
@log_function
|
||||||
def get_stream(self, auth_user_id, pagin_config, timeout=0,
|
def get_stream(self, auth_user_id, pagin_config, timeout=0,
|
||||||
as_client_event=True):
|
as_client_event=True, affect_presence=True):
|
||||||
auth_user = self.hs.parse_userid(auth_user_id)
|
auth_user = self.hs.parse_userid(auth_user_id)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
if auth_user not in self._streams_per_user:
|
if affect_presence and auth_user not in self._streams_per_user:
|
||||||
self._streams_per_user[auth_user] = 0
|
self._streams_per_user[auth_user] = 0
|
||||||
if auth_user in self._stop_timer_per_user:
|
if auth_user in self._stop_timer_per_user:
|
||||||
try:
|
try:
|
||||||
|
@ -64,7 +64,7 @@ class EventStreamHandler(BaseHandler):
|
||||||
yield self.distributor.fire(
|
yield self.distributor.fire(
|
||||||
"started_user_eventstream", auth_user
|
"started_user_eventstream", auth_user
|
||||||
)
|
)
|
||||||
self._streams_per_user[auth_user] += 1
|
self._streams_per_user[auth_user] += 1
|
||||||
|
|
||||||
if pagin_config.from_token is None:
|
if pagin_config.from_token is None:
|
||||||
pagin_config.from_token = None
|
pagin_config.from_token = None
|
||||||
|
@ -92,28 +92,29 @@ class EventStreamHandler(BaseHandler):
|
||||||
defer.returnValue(chunk)
|
defer.returnValue(chunk)
|
||||||
|
|
||||||
finally:
|
finally:
|
||||||
self._streams_per_user[auth_user] -= 1
|
if affect_presence:
|
||||||
if not self._streams_per_user[auth_user]:
|
self._streams_per_user[auth_user] -= 1
|
||||||
del self._streams_per_user[auth_user]
|
if not self._streams_per_user[auth_user]:
|
||||||
|
del self._streams_per_user[auth_user]
|
||||||
|
|
||||||
# 10 seconds of grace to allow the client to reconnect again
|
# 10 seconds of grace to allow the client to reconnect again
|
||||||
# before we think they're gone
|
# before we think they're gone
|
||||||
def _later():
|
def _later():
|
||||||
logger.debug(
|
logger.debug(
|
||||||
"_later stopped_user_eventstream %s", auth_user
|
"_later stopped_user_eventstream %s", auth_user
|
||||||
|
)
|
||||||
|
|
||||||
|
self._stop_timer_per_user.pop(auth_user, None)
|
||||||
|
|
||||||
|
yield self.distributor.fire(
|
||||||
|
"stopped_user_eventstream", auth_user
|
||||||
|
)
|
||||||
|
|
||||||
|
logger.debug("Scheduling _later: for %s", auth_user)
|
||||||
|
self._stop_timer_per_user[auth_user] = (
|
||||||
|
self.clock.call_later(30, _later)
|
||||||
)
|
)
|
||||||
|
|
||||||
self._stop_timer_per_user.pop(auth_user, None)
|
|
||||||
|
|
||||||
yield self.distributor.fire(
|
|
||||||
"stopped_user_eventstream", auth_user
|
|
||||||
)
|
|
||||||
|
|
||||||
logger.debug("Scheduling _later: for %s", auth_user)
|
|
||||||
self._stop_timer_per_user[auth_user] = (
|
|
||||||
self.clock.call_later(30, _later)
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class EventHandler(BaseHandler):
|
class EventHandler(BaseHandler):
|
||||||
|
|
||||||
|
|
|
@ -81,7 +81,9 @@ class Pusher(object):
|
||||||
from_tok = StreamToken.from_string(self.last_token)
|
from_tok = StreamToken.from_string(self.last_token)
|
||||||
config = PaginationConfig(from_token=from_tok, limit='1')
|
config = PaginationConfig(from_token=from_tok, limit='1')
|
||||||
chunk = yield self.evStreamHandler.get_stream(
|
chunk = yield self.evStreamHandler.get_stream(
|
||||||
self.user_name, config, timeout=100*365*24*60*60*1000)
|
self.user_name, config,
|
||||||
|
timeout=100*365*24*60*60*1000, affect_presence=False
|
||||||
|
)
|
||||||
|
|
||||||
# limiting to 1 may get 1 event plus 1 presence event, so
|
# limiting to 1 may get 1 event plus 1 presence event, so
|
||||||
# pick out the actual event
|
# pick out the actual event
|
||||||
|
|
Loading…
Reference in New Issue