Matrix/synapse
Matrix/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse synced 2024-07-02 11:13:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop' into notifier_unify

Browse source
This commit is contained in:
Mark Haines 2015-05-12 11:41:53 +01:00
parent e269c511f6 51da995806
commit 5e4ba463b7
2 changed files with 27 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
synapse/storage/events.py
View file

@ -23,6 +23,7 @@ from synapse.crypto.event_signing import compute_event_reference_hash
from syutil.base64util import decode_base64
from syutil.jsonutil import encode_canonical_json
from contextlib import contextmanager
import logging
@ -41,17 +42,25 @@ class EventsStore(SQLBaseStore):
self.min_token -= 1
stream_ordering = self.min_token
if stream_ordering is None:
stream_ordering_manager = yield self._stream_id_gen.get_next(self)
else:
@contextmanager
def stream_ordering_manager():
yield stream_ordering
try:
yield self.runInteraction(
"persist_event",
self._persist_event_txn,
event=event,
context=context,
backfilled=backfilled,
stream_ordering=stream_ordering,
is_new_state=is_new_state,
current_state=current_state,
)
with stream_ordering_manager as stream_ordering:
yield self.runInteraction(
"persist_event",
self._persist_event_txn,
event=event,
context=context,
backfilled=backfilled,
stream_ordering=stream_ordering,
is_new_state=is_new_state,
current_state=current_state,
)
except _RollbackButIsFineException:
pass
@ -95,15 +104,6 @@ class EventsStore(SQLBaseStore):
# Remove the any existing cache entries for the event_id
txn.call_after(self._invalidate_get_event_cache, event.event_id)
if stream_ordering is None:
with self._stream_id_gen.get_next_txn(txn) as stream_ordering:
return self._persist_event_txn(
txn, event, context, backfilled,
stream_ordering=stream_ordering,
is_new_state=is_new_state,
current_state=current_state,
)
# We purposefully do this first since if we include a `current_state`
# key, we *want* to update the `current_state_events` table
if current_state:

12
synapse/storage/util/id_generators.py
View file

@ -78,14 +78,18 @@ class StreamIdGenerator(object):
self._current_max = None
self._unfinished_ids = deque()
def get_next_txn(self, txn):
@defer.inlineCallbacks
def get_next(self, store):
"""
Usage:
with stream_id_gen.get_next_txn(txn) as stream_id:
with yield stream_id_gen.get_next as stream_id:
# ... persist event ...
"""
if not self._current_max:
self._get_or_compute_current_max(txn)
yield store.runInteraction(
"_compute_current_max",
self._get_or_compute_current_max,
)
with self._lock:
self._current_max += 1
@ -101,7 +105,7 @@ class StreamIdGenerator(object):
with self._lock:
self._unfinished_ids.remove(next_id)
return manager()
defer.returnValue(manager())
@defer.inlineCallbacks
def get_max_token(self, store):