diff --git a/synapse/streams/config.py b/synapse/streams/config.py
index ca7c16ff65..fdda21d165 100644
--- a/synapse/streams/config.py
+++ b/synapse/streams/config.py
@@ -12,11 +12,15 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-
import logging
+from typing import Optional
+
+import attr
from synapse.api.errors import SynapseError
from synapse.http.servlet import parse_integer, parse_string
+from synapse.http.site import SynapseRequest
+from synapse.storage.databases.main import DataStore
from synapse.types import StreamToken
logger = logging.getLogger(__name__)
@@ -25,38 +29,23 @@ logger = logging.getLogger(__name__)
MAX_LIMIT = 1000
-class SourcePaginationConfig(object):
-
- """A configuration object which stores pagination parameters for a
- specific event source."""
-
- def __init__(self, from_key=None, to_key=None, direction="f", limit=None):
- self.from_key = from_key
- self.to_key = to_key
- self.direction = "f" if direction == "f" else "b"
- self.limit = min(int(limit), MAX_LIMIT) if limit is not None else None
-
- def __repr__(self):
- return "StreamConfig(from_key=%r, to_key=%r, direction=%r, limit=%r)" % (
- self.from_key,
- self.to_key,
- self.direction,
- self.limit,
- )
-
-
-class PaginationConfig(object):
-
+@attr.s(slots=True)
+class PaginationConfig:
"""A configuration object which stores pagination parameters."""
- def __init__(self, from_token=None, to_token=None, direction="f", limit=None):
- self.from_token = from_token
- self.to_token = to_token
- self.direction = "f" if direction == "f" else "b"
- self.limit = min(int(limit), MAX_LIMIT) if limit is not None else None
+ from_token = attr.ib(type=Optional[StreamToken])
+ to_token = attr.ib(type=Optional[StreamToken])
+ direction = attr.ib(type=str)
+ limit = attr.ib(type=Optional[int])
@classmethod
- def from_request(cls, request, raise_invalid_params=True, default_limit=None):
+ async def from_request(
+ cls,
+ store: "DataStore",
+ request: SynapseRequest,
+ raise_invalid_params: bool = True,
+ default_limit: Optional[int] = None,
+ ) -> "PaginationConfig":
direction = parse_string(request, "dir", default="f", allowed_values=["f", "b"])
from_tok = parse_string(request, "from")
@@ -66,20 +55,23 @@ class PaginationConfig(object):
if from_tok == "END":
from_tok = None # For backwards compat.
elif from_tok:
- from_tok = StreamToken.from_string(from_tok)
+ from_tok = await StreamToken.from_string(store, from_tok)
except Exception:
raise SynapseError(400, "'from' parameter is invalid")
try:
if to_tok:
- to_tok = StreamToken.from_string(to_tok)
+ to_tok = await StreamToken.from_string(store, to_tok)
except Exception:
raise SynapseError(400, "'to' parameter is invalid")
limit = parse_integer(request, "limit", default=default_limit)
- if limit and limit < 0:
- raise SynapseError(400, "Limit must be 0 or above")
+ if limit:
+ if limit < 0:
+ raise SynapseError(400, "Limit must be 0 or above")
+
+ limit = min(int(limit), MAX_LIMIT)
try:
return PaginationConfig(from_tok, to_tok, direction, limit)
@@ -87,20 +79,10 @@ class PaginationConfig(object):
logger.exception("Failed to create pagination config")
raise SynapseError(400, "Invalid request.")
- def __repr__(self):
+ def __repr__(self) -> str:
return ("PaginationConfig(from_tok=%r, to_tok=%r, direction=%r, limit=%r)") % (
self.from_token,
self.to_token,
self.direction,
self.limit,
)
-
- def get_source_config(self, source_name):
- keyname = "%s_key" % source_name
-
- return SourcePaginationConfig(
- from_key=getattr(self.from_token, keyname),
- to_key=getattr(self.to_token, keyname) if self.to_token else None,
- direction=self.direction,
- limit=self.limit,
- )
diff --git a/synapse/streams/events.py b/synapse/streams/events.py
index 5d3eddcfdc..92fd5d489f 100644
--- a/synapse/streams/events.py
+++ b/synapse/streams/events.py
@@ -15,8 +15,6 @@
from typing import Any, Dict
-from twisted.internet import defer
-
from synapse.handlers.account_data import AccountDataEventSource
from synapse.handlers.presence import PresenceEventSource
from synapse.handlers.receipts import ReceiptEventSource
@@ -25,7 +23,7 @@ from synapse.handlers.typing import TypingNotificationEventSource
from synapse.types import StreamToken
-class EventSources(object):
+class EventSources:
SOURCE_TYPES = {
"room": RoomEventSource,
"presence": PresenceEventSource,
@@ -40,19 +38,18 @@ class EventSources(object):
} # type: Dict[str, Any]
self.store = hs.get_datastore()
- @defer.inlineCallbacks
- def get_current_token(self):
- push_rules_key, _ = self.store.get_push_rules_stream_token()
+ def get_current_token(self) -> StreamToken:
+ push_rules_key = self.store.get_max_push_rules_stream_id()
to_device_key = self.store.get_to_device_stream_token()
device_list_key = self.store.get_device_stream_token()
groups_key = self.store.get_group_stream_token()
token = StreamToken(
- room_key=(yield self.sources["room"].get_current_key()),
- presence_key=(yield self.sources["presence"].get_current_key()),
- typing_key=(yield self.sources["typing"].get_current_key()),
- receipt_key=(yield self.sources["receipt"].get_current_key()),
- account_data_key=(yield self.sources["account_data"].get_current_key()),
+ room_key=self.sources["room"].get_current_key(),
+ presence_key=self.sources["presence"].get_current_key(),
+ typing_key=self.sources["typing"].get_current_key(),
+ receipt_key=self.sources["receipt"].get_current_key(),
+ account_data_key=self.sources["account_data"].get_current_key(),
push_rules_key=push_rules_key,
to_device_key=to_device_key,
device_list_key=device_list_key,
@@ -60,8 +57,7 @@ class EventSources(object):
)
return token
- @defer.inlineCallbacks
- def get_current_token_for_pagination(self):
+ def get_current_token_for_pagination(self) -> StreamToken:
"""Get the current token for a given room to be used to paginate
events.
@@ -69,10 +65,10 @@ class EventSources(object):
than `room`, since they are not used during pagination.
Returns:
- Deferred[StreamToken]
+ The current token for pagination.
"""
token = StreamToken(
- room_key=(yield self.sources["room"].get_current_key()),
+ room_key=self.sources["room"].get_current_key(),
presence_key=0,
typing_key=0,
receipt_key=0,
|
