Repository: hikari-py/hikari
Pull request #906 (2021-11-15 19:16:05)

Switch to more relaxed requirements


Patch:
diff --git a/changes/864.deprecation.md b/changes/864.deprecation.md
new file mode 100644
index 00000000..5e274d53
--- /dev/null
+++ b/changes/864.deprecation.md
@@ -0,0 +1,1 @@
+EventStream is now a sync context menu not async.
diff --git a/changes/864.feature.md b/changes/864.feature.md
new file mode 100644
index 00000000..6e7fd063
--- /dev/null
+++ b/changes/864.feature.md
@@ -0,0 +1,1 @@
+`EventStream.filter` now always returns `EventStream`.
diff --git a/changes/897.breaking.md b/changes/897.breaking.md
new file mode 100644
index 00000000..76259a50
--- /dev/null
+++ b/changes/897.breaking.md
@@ -0,0 +1,2 @@
+Split bulk message delete from normal delete
+  - The new event is now `hikari.events.message_events.GuildBulkMessageDeleteEvent`
diff --git a/changes/897.feature.md b/changes/897.feature.md
new file mode 100644
index 00000000..00052b25
--- /dev/null
+++ b/changes/897.feature.md
@@ -0,0 +1,1 @@
+Add `old_message` attribute to `hikari.events.message_events.MessageDelete`
diff --git a/changes/906.feature.md b/changes/906.feature.md
new file mode 100644
index 00000000..614e896b
--- /dev/null
+++ b/changes/906.feature.md
@@ -0,0 +1,1 @@
+Switch to more relaxed requirements.
diff --git a/hikari/api/event_factory.py b/hikari/api/event_factory.py
index 8f108055..3a58e196 100644
--- a/hikari/api/event_factory.py
+++ b/hikari/api/event_factory.py
@@ -36,6 +36,7 @@
     from hikari import invites as invite_models
     from hikari import messages as messages_models
     from hikari import presences as presences_models
+    from hikari import snowflakes
     from hikari import users as user_models
     from hikari import voices as voices_models
     from hikari.api import shard as gateway_shard
@@ -752,7 +753,11 @@ def deserialize_message_update_event(
 
     @abc.abstractmethod
     def deserialize_message_delete_event(
-        self, shard: gateway_shard.GatewayShard, payload: data_binding.JSONObject
+        self,
+        shard: gateway_shard.GatewayShard,
+        payload: data_binding.JSONObject,
+        *,
+        old_message: typing.Optional[messages_models.Message],
     ) -> message_events.MessageDeleteEvent:
         """Parse a raw payload from Discord into a message delete event object.
 
@@ -762,6 +767,8 @@ def deserialize_message_delete_event(
             The shard that emitted this event.
         payload : hikari.internal.data_binding.JSONObject
             The dict payload to parse.
+        old_message : typing.Optional[hikari.messages.Message]
+            The old message object.
 
         Returns
         -------
@@ -770,10 +777,14 @@ def deserialize_message_delete_event(
         """
 
     @abc.abstractmethod
-    def deserialize_message_delete_bulk_event(
-        self, shard: gateway_shard.GatewayShard, payload: data_binding.JSONObject
-    ) -> message_events.MessageDeleteEvent:
-        """Parse a raw payload from Discord into a message delete bulk event object.
+    def deserialize_guild_message_delete_bulk_event(
+        self,
+        shard: gateway_shard.GatewayShard,
+        payload: data_binding.JSONObject,
+        *,
+        old_messages: typing.Mapping[snowflakes.Snowflake, messages_models.Message],
+    ) -> message_events.GuildBulkMessageDeleteEvent:
+        """Parse a raw payload from Discord into a guild message delete bulk event object.
 
         Parameters
         ----------
@@ -781,16 +792,13 @@ def deserialize_message_delete_bulk_event(
             The shard that emitted this event.
         payload : hikari.internal.data_binding.JSONObject
             The dict payload to parse.
+        old_messages : typing.Mapping[hikari.snowflakes.Snowflake, hikari.messages.Message]
+            A mapping of the old message objects.
 
         Returns
         -------
-        hikari.events.message_events.MessageDeleteEvent
-            The parsed message delete bulk event object.
-
-        Raises
-        ------
-        builtins.NotImplementedError
-            If a bulk delete occurs in a DM channel.
+        hikari.events.message_events.GuildBulkMessageDeleteEvent
+            The parsed guild message delete bulk event object.
         """
 
     ###################
diff --git a/hikari/api/event_manager.py b/hikari/api/event_manager.py
index 90ec286c..8cd02c76 100644
--- a/hikari/api/event_manager.py
+++ b/hikari/api/event_manager.py
@@ -57,12 +57,12 @@ class EventStream(iterators.LazyIterator[EventT], abc.ABC):
 
     Examples
     --------
-    A streamer may either be started and closed using `async with` syntax
+    A streamer may either be started and closed using `with` syntax
     where `EventStream.open` and `EventStream.close` are implicitly called based on
     context.
 
     ```py
-    async with EventStream(app, EventType, timeout=50) as stream:
+    with EventStream(app, EventType, timeout=50) as stream:
         async for entry in stream:
             ...
     ```
@@ -89,33 +89,33 @@ class EventStream(iterators.LazyIterator[EventT], abc.ABC):
     __slots__: typing.Sequence[str] = ()
 
     @abc.abstractmethod
-    async def close(self) -> None:
+    def close(self) -> None:
         """Mark this streamer as closed to stop it from queueing and receiving events.
 
         If called on an already closed streamer then this will do nothing.
 
         !!! note
-            `async with streamer` may be used as a short-cut for opening and
+            `with streamer` may be used as a short-cut for opening and
             closing a streamer.
         """
 
     @abc.abstractmethod
-    async def open(self) -> None:
+    def open(self) -> None:
         """Mark this streamer as opened to let it start receiving and queueing events.
 
         If called on an already started streamer then this will do nothing.
 
         !!! note
-            `async with streamer` may be used as a short-cut for opening and
+            `with streamer` may be used as a short-cut for opening and
             closing a stream.
         """
 
     @abc.abstractmethod
-    async def __aenter__(self) -> EventStream[EventT]:
+    def __enter__(self) -> EventStream[EventT]:
         raise NotImplementedError
 
     @abc.abstractmethod
-    async def __aexit__(
+    def __exit__(
         self,
         exc_type: typing.Optional[typing.Type[BaseException]],
         exc: typing.Optional[BaseException],
@@ -416,33 +416,33 @@ def stream(
         -------
         EventStream[hikari.events.base_events.Event]
             The async iterator to handle streamed events. This must be started
-            with `async with stream:` or `await stream.open()` before
+            with `with stream:` or `stream.open()` before
             asynchronously iterating over it.
 
         !!! warning
-            If you use `await stream.open()` to start the stream then you must
-            also close it with `await stream.close()` otherwise it may queue
+            If you use `stream.open()` to start the stream then you must
+            also close it with `stream.close()` otherwise it may queue
             events in memory indefinitely.
 
         Examples
         --------
 
         ```py
-        async with bot.stream(events.ReactionAddEvent, timeout=30).filter(("message_id", message.id)) as stream:
+        with bot.stream(events.ReactionAddEvent, timeout=30).filter(("message_id", message.id)) as stream:
             async for user_id in stream.map("user_id").limit(50):
                 ...
         ```
 
-        or using await `open()` and await `close()`
+        or using `open()` and `close()`
 
         ```py
         stream = bot.stream(events.ReactionAddEvent, timeout=30).filter(("message_id", message.id))
-        await stream.open()
+        stream.open()
 
         async for user_id in stream.map("user_id").limit(50)
             ...
 
-        await stream.close()
+        stream.close()
         ```
 
         See Also
diff --git a/hikari/events/message_events.py b/hikari/events/message_events.py
index 67f50c52..31c23186 100644
--- a/hikari/events/message_events.py
+++ b/hikari/events/message_events.py
@@ -32,6 +32,7 @@
     "GuildMessageCreateEvent",
     "GuildMessageUpdateEvent",
     "GuildMessageDeleteEvent",
+    "GuildBulkMessageDeleteEvent",
     "DMMessageCreateEvent",
     "DMMessageUpdateEvent",
     "DMMessageDeleteEvent",
@@ -594,57 +595,26 @@ class DMMessageUpdateEvent(MessageUpdateEvent):
 
 @base_events.requires_intents(intents.Intents.GUILD_MESSAGES, intents.Intents.DM_MESSAGES)
 class MessageDeleteEvent(MessageEvent, abc.ABC):
-    """Special event that is triggered when one or more messages get deleted.
+    """Special event that is triggered when a message gets deleted.
 
     !!! note
         Due to Discord limitations, most message information is unavailable
         during deletion events.
-
-    You can check if the message was in a singular deletion by checking the
-    `is_bulk` attribute.
     """
 
     __slots__: typing.Sequence[str] = ()
 
-    @property
-    def message_id(self) -> snowflakes.Snowflake:
-        """Get the ID of the first deleted message.
-
-        This is contextually useful if you know this is not a bulk deletion
-        event. For all other purposes, this is the same as running
-        `next(iter(event.message_ids))`.
-
-        Returns
-        -------
-        hikari.snowflakes.Snowflake
-            The first deleted message ID.
-        """
-        try:
-            return next(iter(self.message_ids))
-        except StopIteration:
-            raise RuntimeError("No messages were sent in a bulk delete! Please shout at Discord to fix this!") from None
-
     @property
     @abc.abstractmethod
-    def message_ids(self) -> typing.AbstractSet[snowflakes.Snowflake]:
-        """Set of message IDs that were bulk deleted.
-
-        Returns
-        -------
-        typing.AbstractSet[hikari.snowflakes.Snowflake]
-            A sequence of message IDs that were bulk deleted.
-        """
+    def message_id(self) -> snowflakes.Snowflake:
+        """ID of the message that was deleted."""
 
     @property
     @abc.abstractmethod
-    def is_bulk(self) -> bool:
-        """Flag that determines whether this was a bulk deletion or not.
+    def old_message(self) -> typing.Optional[messages.Message]:
+        """Object of the message that was deleted.
 
-        Returns
-        -------
-        builtins.bool
-            `builtins.True` if this was a bulk deletion, or `builtins.False`
-            if it was a regular message deletion.
+        Will be `None` if the message was not found in the cache.
         """
 
 
@@ -652,15 +622,11 @@ def is_bulk(self) -> bool:
 @attr.define(kw_only=True, weakref_slot=False)
 @base_events.requires_intents(intents.Intents.GUILD_MESSAGES)
 class GuildMessageDeleteEvent(MessageDeleteEvent):
-    """Event that is triggered if messages are deleted in a guild.
+    """Event that is triggered if a message is deleted in a guild.
 
     !!! note
         Due to Discord limitations, most message information is unavailable
         during deletion events.
-
-    This is triggered for singular message deletion, and bulk message
-    deletion. You can check if the message was in a singular deletion by
-    checking the `is_bulk` attribute.
     """
 
     app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True})
@@ -670,25 +636,19 @@ class GuildMessageDeleteEvent(MessageDeleteEvent):
     # <<inherited docstring from MessageEvent>>
 
     guild_id: snowflakes.Snowflake = attr.field()
-    """ID of the guild that this event occurred in.
+    """ID of the guild that this event occurred in."""
 
-    Returns
-    -------
-    hikari.snowflakes.Snowflake
-        The ID of the guild.
-    """
-
-    is_bulk: bool = attr.field()
+    message_id: snowflakes.Snowflake = attr.field()
     # <<inherited docstring from MessageDeleteEvent>>
 
-    message_ids: typing.AbstractSet[snowflakes.Snowflake] = attr.field()
+    old_message: typing.Optional[messages.Message] = attr.field()
     # <<inherited docstring from MessageDeleteEvent>>
 
     shard: shard_.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True})
     # <<inherited docstring from ShardEvent>>
 
     def get_channel(self) -> typing.Optional[channels.TextableGuildChannel]:
-        """Get the cached channel the messages were sent in, if known.
+        """Get the cached channel the message were sent in, if known.
 
         Returns
         -------
@@ -728,17 +688,11 @@ def get_guild(self) -> typing.Optional[guilds.GatewayGuild]:
 @attr.define(kw_only=True, weakref_slot=False)
 @base_events.requires_intents(intents.Intents.DM_MESSAGES)
 class DMMessageDeleteEvent(MessageDeleteEvent):
-    """Event that is triggered if messages are deleted in a DM.
+    """Event that is triggered if a message is deleted in a DM.
 
     !!! note
         Due to Discord limitations, most message information is unavailable
         during deletion events.
-
-    This is triggered for singular message deletion, and bulk message
-    deletion, although the latter is not expected to occur in DMs.
-
-    You can check if the message was in a singular deletion by checking the
-    `is_bulk` attribute.
     """
 
     app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True})
@@ -747,11 +701,80 @@ class DMMessageDeleteEvent(MessageDeleteEvent):
     channel_id: snowflakes.Snowflake = attr.field()
     # <<inherited docstring from MessageEvent>>
 
-    is_bulk: bool = attr.field()
+    message_id: snowflakes.Snowflake = attr.field()
     # <<inherited docstring from MessageDeleteEvent>>
 
-    message_ids: typing.AbstractSet[snowflakes.Snowflake] = attr.field()
+    old_message: typing.Optional[messages.Message] = attr.field()
     # <<inherited docstring from MessageDeleteEvent>>
 
     shard: shard_.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True})
     # <<inherited docstring from ShardEvent>>
+
+
+@attr_extensions.with_copy
+@attr.define(kw_only=True, weakref_slot=False)
+@base_events.requires_intents(intents.Intents.GUILD_MESSAGES)
+class GuildBulkMessageDeleteEvent(shard_events.ShardEvent):
+    """Event that is triggered when a bulk deletion is triggered in a guild.
+
+    !!! note
+        Due to Discord limitations, most message information is unavailable
+        during deletion events.
+    """
+
+    app: traits.RESTAware = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True})
+    # <<inherited docstring from Event>>
+
+    channel_id: snowflakes.Snowflake = attr.field()
+    """ID of the channel that this event concerns."""
+
+    guild_id: snowflakes.Snowflake = attr.field()
+    """ID of the guild that this event occurred in."""
+
+    message_ids: typing.AbstractSet[snowflakes.Snowflake] = attr.field()
+    """Set of message IDs that were bulk deleted."""
+
+    old_messages: typing.Mapping[snowflakes.Snowflake, messages.Message] = attr.field()
+    """Mapping of a snowflake to the deleted message object.
+
+    If the message was not found in the cache it will be missing from the mapping.
+    """
+
+    shard: shard_.GatewayShard = attr.field(metadata={attr_extensions.SKIP_DEEP_COPY: True})
+    # <<inherited docstring from ShardEvent>>
+
+    def get_channel(self) -> typing.Optional[channels.TextableGuildChannel]:
+        """Get the cached channel the messages were sent in, if known.
+
+        Returns
+        -------
+        typing.Optional[hikari.channels.TextableGuildChannel]
+            The channel the messages were sent in, or `builtins.None` if not
+            known/cached.
+        """
+        if not isinstance(self.app, traits.CacheAware):
+            return None
+
+        channel = self.app.cache.get_guild_channel(self.channel_id)
+        assert channel is None or isinstance(
+            channel, channels.TextableGuildChannel
+        ), f"Cached channel ID is not a TextableGuildChannel, but a {type(channel).__name__}!"
+        return channel
+
+    def get_guild(self) -> typing.Optional[guilds.GatewayGuild]:
+        """Get the cached guild this event corresponds to, if known.
+
+        !!! note
+            You will need `hikari.intents.Intents.GUILDS` enabled to receive this
+            information.
+
+        Returns
+        -------
+        hikari.guilds.GatewayGuild
+            The gateway guild that this event corresponds to, if known and
+            cached.
+        """
+        if not isinstance(self.app, traits.CacheAware):
+            return None
+
+        return self.app.cache.get_guild(self.guild_id)
diff --git a/hikari/impl/event_factory.py b/hikari/impl/event_factory.py
index 92b6917c..99fa7c6d 100644
--- a/hikari/impl/event_factory.py
+++ b/hikari/impl/event_factory.py
@@ -480,7 +480,7 @@ def deserialize_message_create_event(
 
         return message_events.GuildMessageCreateEvent(shard=shard, message=message)
 
-    def deserialize_message_update_event(  # noqa: CFQ001
+    def deserialize_message_update_event(
         self,
         shard: gateway_shard.GatewayShard,
         payload: data_binding.JSONObject,
@@ -495,48 +495,46 @@ def deserialize_message_update_event(  # noqa: CFQ001
         return message_events.GuildMessageUpdateEvent(shard=shard, message=message, old_message=old_message)
 
     def deserialize_message_delete_event(
-        self, shard: gateway_shard.GatewayShard, payload: data_binding.JSONObject
+        self,
+        shard: gateway_shard.GatewayShard,
+        payload: data_binding.JSONObject,
+        *,
+        old_message: typing.Optional[messages_models.Message],
     ) -> message_events.MessageDeleteEvent:
         channel_id = snowflakes.Snowflake(payload["channel_id"])
-        message_ids = collections.SnowflakeSet(int(payload["id"]))
+        message_id = snowflakes.Snowflake(payload["id"])
 
         if "guild_id" in payload:
             return message_events.GuildMessageDeleteEvent(
                 app=self._app,
                 shard=shard,
                 channel_id=channel_id,
-                message_ids=message_ids,
-                is_bulk=False,
+                message_id=message_id,
                 guild_id=snowflakes.Snowflake(payload["guild_id"]),
+                old_message=old_message,
             )
 
         return message_events.DMMessageDeleteEvent(
             app=self._app,
             shard=shard,
             channel_id=channel_id,
-            message_ids=message_ids,
-            is_bulk=False,
+            message_id=message_id,
+            old_message=old_message,
         )
 
-    def deserialize_message_delete_bulk_event(
-        self, shard: gateway_shard.GatewayShard, payload: data_binding.JSONObject
-    ) -> message_events.MessageDeleteEvent:
-
-        message_ids = collections.SnowflakeSet(*(snowflakes.Snowflake(message_id) for message_id in payload["ids"]))
-        channel_id = snowflakes.Snowflake(payload["channel_id"])
-
-        if "guild_id" in payload:
-            return message_events.GuildMessageDeleteEvent(
-                app=self._app,
-                shard=shard,
-                channel_id=channel_id,
-                guild_id=snowflakes.Snowflake(payload["guild_id"]),
-                message_ids=message_ids,
-                is_bulk=True,
-            )
-
-        return message_events.DMMessageDeleteEvent(
-            app=self._app, shard=shard, channel_id=channel_id, message_ids=message_ids, is_bulk=True
+    def deserialize_guild_message_delete_bulk_event(
+        self,
+        shard: gateway_shard.GatewayShard,
+        payload: data_binding.JSONObject,
+        old_messages: typing.Mapping[snowflakes.Snowflake, messages_models.Message],
+    ) -> message_events.GuildBulkMessageDeleteEvent:
+        return message_events.GuildBulkMessageDeleteEvent(
+            app=self._app,
+            shard=shard,
+            channel_id=snowflakes.Snowflake(payload["channel_id"]),
+            guild_id=snowflakes.Snowflake(payload["guild_id"]),
+            message_ids=collections.SnowflakeSet(*(snowflakes.Snowflake(message_id) for message_id in payload["ids"])),
+            old_messages=old_messages,
         )
 
     ###################
diff --git a/hikari/impl/event_manager.py b/hikari/impl/event_manager.py
index bcd72d9e..25c43255 100644
--- a/hikari/impl/event_manager.py
+++ b/hikari/impl/event_manager.py
@@ -389,22 +389,30 @@ async def on_message_update(self, shard: gateway_shard.GatewayShard, payload: da
 
     async def on_message_delete(self, shard: gateway_shard.GatewayShard, payload: data_binding.JSONObject) -> None:
         """See https://discord.com/developers/docs/topics/gateway#message-delete for more info."""
-        event = self._event_factory.deserialize_message_delete_event(shard, payload)
-
         if self._cache:
-            self._cache.delete_message(event.message_id)
+            message_id = snowflakes.Snowflake(payload["id"])
+            old_message = self._cache.delete_message(message_id)
+        else:
+            old_message = None
+
+        event = self._event_factory.deserialize_message_delete_event(shard, payload, old_message=old_message)
 
         await self.dispatch(event)
 
     async def on_message_delete_bulk(self, shard: gateway_shard.GatewayShard, payload: data_binding.JSONObject) -> None:
         """See https://discord.com/developers/docs/topics/gateway#message-delete-bulk for more info."""
-        event = self._event_factory.deserialize_message_delete_bulk_event(shard, payload)
+        old_messages = {}
 
         if self._cache:
-            for message_id in event.message_ids:
-                self._cache.delete_message(message_id)
+            for message_id in payload["ids"]:
+                message_id = snowflakes.Snowflake(message_id)
 
-        await self.dispatch(event)
+                if message := self._cache.delete_message(message_id):
+                    old_messages[message_id] = message
+
+        await self.dispatch(
+            self._event_factory.deserialize_guild_message_delete_bulk_event(shard, payload, old_messages=old_messages)
+        )
 
     async def on_message_reaction_add(
         self, shard: gateway_shard.GatewayShard, payload: data_binding.JSONObject
diff --git a/hikari/impl/event_manager_base.py b/hikari/impl/event_manager_base.py
index 0c66ed64..3e3290a3 100644
--- a/hikari/impl/event_manager_base.py
+++ b/hikari/impl/event_manager_base.py
@@ -64,6 +64,7 @@
     WaiterMapT = typing.MutableMapping[
         typing.Type[event_manager_.EventT_co], typing.MutableSet[WaiterT[event_manager_.EventT_co]]
     ]
+    _EventStreamT = typing.TypeVar("_EventStreamT", bound="EventStream[typing.Any]")
 
 
 def _generate_weak_listener(
@@ -94,9 +95,11 @@ class EventStream(event_manager_.EventStream[event_manager_.EventT]):
     __slots__: typing.Sequence[str] = (
         "__weakref__",
         "_active",
+        "_event",
         "_event_manager",
         "_event_type",
         "_filters",
+        "_limit",
         "_queue",
         "_registered_listener",
         "_timeout",
@@ -110,55 +113,70 @@ def __init__(
         timeout: typing.Union[float, int, None],
         limit: typing.Optional[int] = None,
     ) -> None:
-        self._event_manager = event_manager
         self._active = False
+        self._event: typing.Optional[asyncio.Event] = None
+        self._event_manager = event_manager
         self._event_type = event_type
         self._filters: iterators.All[event_manager_.EventT] = iterators.All(())
-        # We accept `None` to represent unlimited here to be consistent with how `None` is already used to represent
-        # unlimited for timeout in other places.
-        self._queue: asyncio.Queue[event_manager_.EventT] = asyncio.Queue(limit or 0)
+        self._limit = limit
+        self._queue: typing.List[event_manager_.EventT] = []
         self._registered_listener: typing.Optional[
             typing.Callable[[event_manager_.EventT], typing.Coroutine[typing.Any, typing.Any, None]]
         ] = None
         # The registered wrapping function for the weak ref to this class's _listener method.
         self._timeout = timeout
 
-    async def __aenter__(self) -> EventStream[event_manager_.EventT]:
-        await self.open()
+    # These are only included at runtime in-order to avoid the model being typed as an asynchronous context manager.
+    if not typing.TYPE_CHECKING:
+
+        async def __aenter__(self: _EventStreamT) -> _EventStreamT:
+            # This is sync only.
+            warnings.warn(
+                "Using EventStream as an async context manager has been deprecated since 2.0.0.dev104. "
+                "Please use it as a sycnrhonous context manager (e.g. with bot.stream(...)) instead.",
+                category=DeprecationWarning,
+                stacklevel=2,
+            )
+
+            self.open()
+            return self
+
+        async def __aexit__(
+            self,
+            exc_type: typing.Optional[typing.Type[BaseException]],
+            exc: typing.Optional[BaseException],
+            exc_tb: typing.Optional[types.TracebackType],
+        ) -> None:
+            self.close()
+
+    def __enter__(self: _EventStreamT) -> _EventStreamT:
+        self.open()
         return self
 
-    async def __aexit__(
+    def __exit__(
         self,
         exc_type: typing.Optional[typing.Type[BaseException]],
-        exc: typing.Optional[BaseException],
+        exc_val: typing.Optional[BaseException],
         exc_tb: typing.Optional[types.TracebackType],
     ) -> None:
-        await self.close()
+        self.close()
 
-    # These are only included at runtime in-order to avoid the model being typed as a synchronous context manager.
-    if not typing.TYPE_CHECKING:
+    async def __anext__(self) -> event_manager_.EventT:
+        if not self._active:
+            raise TypeError("stream must be started with before entering it")
 
-        def __enter__(self) -> typing.NoReturn:
-            # This is async only.
-            cls = type(self)
-            raise TypeError(f"{cls.__module__}.{cls.__qualname__} is async-only, did you mean 'async with'?") from None
+        while not self._queue:
+            if not self._event:
+                self._event = asyncio.Event()
 
-        def __exit__(
-            self,
-            exc_type: typing.Optional[typing.Type[Exception]],
-            exc_val: typing.Optional[Exception],
-            exc_tb: typing.Optional[types.TracebackType],
-        ) -> None:
-            return None
+            try:
+                await asyncio.wait_for(self._event.wait(), timeout=self._timeout)
+            except asyncio.TimeoutError:
+                raise StopAsyncIteration from None
 
-    async def __anext__(self) -> event_manager_.EventT:
-        if not self._active:
-            raise TypeError("stream must be started with `async with` before entering it")
+            self._event.clear()
 
-        try:
-            return await asyncio.wait_for(self._queue.get(), timeout=self._timeout)
-        except asyncio.TimeoutError:
-            raise StopAsyncIteration from None
+        return self._queue.pop(0)
 
     def __await__(self) -> typing.Generator[None, None, typing.Sequence[event_manager_.EventT]]:
         return self._await_all().__await__()
@@ -170,27 +188,23 @@ def __del__(self) -> None:
         # ominous memory leak.
         if self._active:
             _LOGGER.warning("active %r streamer fell out of scope before being closed", self._event_type.__name__)
-            try:
-                asyncio.ensure_future(self.close())
-            except RuntimeError:
-                pass
+            self.close()
 
     async def _await_all(self) -> typing.Sequence[event_manager_.EventT]:
-        await self.open()
+        self.open()
         result = [event async for event in self]
-        await self.close()
+        self.close()
         return result
 
     async def _listener(self, event: event_manager_.EventT) -> None:
-        if not self._filters(event):
+        if not self._filters(event) or (self._limit is not None and len(self._queue) >= self._limit):
             return
 
-        try:
-            self._queue.put_nowait(event)
-        except asyncio.QueueFull:
-            pass
+        self._queue.append(event)
+        if self._event:
+            self._event.set()
 
-    async def close(self) -> None:
+    def close(self) -> None:
         if self._active and self._registered_listener is not None:
             try:
                 self._event_manager.unsubscribe(self._event_type, self._registered_listener)
@@ -202,17 +216,18 @@ async def close(self) -> None:
         self._active = False
 
     def filter(
-        self,
+        self: _EventStreamT,
         *predicates: typing.Union[typing.Tuple[str, typing.Any], typing.Callable[[event_manager_.EventT], bool]],
         **attrs: typing.Any,
-    ) -> typing.Union[EventStream[event_manager_.EventT], iterators.LazyIterator[event_manager_.EventT]]:
+    ) -> _EventStreamT:
+        filter_ = self._map_predicates_and_attr_getters("filter", *predicates, **attrs)
         if self._active:
-            return super().filter(*predicates, **attrs)
+            self._queue = [entry for entry in self._queue if filter_(entry)]
 
-        self._filters |= self._map_predicates_and_attr_getters("filter", *predicates, **attrs)
+        self._filters |= filter_
         return self
 
-    async def open(self) -> None:
+    def open(self) -> None:
         if not self._active:
             # For the sake of protecting highly intelligent people who forget to close this, we register the event
             # listener with a weakref then try to close this on deletion. While this may lead to their consoles being
diff --git a/hikari/impl/rest.py b/hikari/impl/rest.py
index 4385659e..20806553 100644
--- a/hikari/impl/rest.py
+++ b/hikari/impl/rest.py
@@ -670,8 +670,8 @@ def __enter__(self) -> typing.NoReturn:
 
         def __exit__(
             self,
-            exc_type: typing.Optional[typing.Type[Exception]],
-            exc_val: typing.Optional[Exception],
+            exc_type: typing.Optional[typing.Type[BaseException]],
+            exc_val: typing.Optional[BaseException],
             exc_tb: typing.Optional[types.TracebackType],
         ) -> None:
             return None
diff --git a/requirements.txt b/requirements.txt
index 1fa5f195..16ce2bf9 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,5 +1,5 @@
-aiohttp==3.8.1
-attrs==21.2.0
-colorlog==6.6.0
-multidict==5.2.0
+aiohttp~=3.8
+attrs~=21.2
+colorlog~=6.6
+multidict~=5.2
 pure25519==0.0.1
diff --git a/speedup-requirements.txt b/speedup-requirements.txt
index fc51178f..c2cc48b7 100644
--- a/speedup-requirements.txt
+++ b/speedup-requirements.txt
@@ -1,5 +1,5 @@
-aiodns==3.0.0
-cchardet==2.1.7
-Brotli==1.0.9
-ciso8601==2.2.0
-ed25519==1.5
+aiodns~=3.0
+cchardet~=2.1
+Brotli~=1.0
+ciso8601~=2.2
+ed25519~=1.5


Test patch:
diff --git a/tests/hikari/events/test_message_events.py b/tests/hikari/events/test_message_events.py
index e59bea4a..5476efa2 100644
--- a/tests/hikari/events/test_message_events.py
+++ b/tests/hikari/events/test_message_events.py
@@ -319,17 +319,6 @@ def test_old_message(self, event):
         assert event.old_message.id == 123
 
 
-class TestMessageDeleteEvent:
-    def test_message_id_property(self):
-        event = hikari_test_helpers.mock_class_namespace(message_events.MessageDeleteEvent, message_ids=[9, 18, 27])()
-        assert event.message_id == 9
-
-    def test_message_id_property_if_empty_errors(self):
-        event = hikari_test_helpers.mock_class_namespace(message_events.MessageDeleteEvent, message_ids=[])()
-        with pytest.raises(RuntimeError):
-            _ = event.message_id
-
-
 class TestGuildMessageDeleteEvent:
     @pytest.fixture()
     def event(self):
@@ -337,9 +326,9 @@ def event(self):
             guild_id=snowflakes.Snowflake(542342354564),
             channel_id=snowflakes.Snowflake(54213123123),
             app=mock.Mock(),
-            shard=mock.Mock(),
-            message_ids={9, 18, 27, 36},
-            is_bulk=True,
+            shard=object(),
+            message_id=9,
+            old_message=object(),
         )
 
     def test_get_channel_when_no_cache_trait(self, event):
diff --git a/tests/hikari/impl/test_event_factory.py b/tests/hikari/impl/test_event_factory.py
index a601fa0d..eece3e08 100644
--- a/tests/hikari/impl/test_event_factory.py
+++ b/tests/hikari/impl/test_event_factory.py
@@ -615,54 +615,47 @@ def test_deserialize_message_update_event_in_dm(self, event_factory, mock_app, m
 
     def test_deserialize_message_delete_event_in_guild(self, event_factory, mock_app, mock_shard):
         mock_payload = {"id": "5412", "channel_id": "541123", "guild_id": "9494949"}
+        old_message = object()
 
-        event = event_factory.deserialize_message_delete_event(mock_shard, mock_payload)
+        event = event_factory.deserialize_message_delete_event(mock_shard, mock_payload, old_message=old_message)
 
         assert isinstance(event, message_events.GuildMessageDeleteEvent)
         assert event.app is mock_app
         assert event.shard is mock_shard
+        assert event.old_message is old_message
         assert event.channel_id == 541123
-        assert event.message_ids == {5412}
-        assert event.is_bulk is False
+        assert event.message_id == 5412
         assert event.guild_id == 9494949
 
     def test_deserialize_message_delete_event_in_dm(self, event_factory, mock_app, mock_shard):
         mock_payload = {"id": "5412", "channel_id": "541123"}
+        old_message = object()
 
-        event = event_factory.deserialize_message_delete_event(mock_shard, mock_payload)
+        event = event_factory.deserialize_message_delete_event(mock_shard, mock_payload, old_message=old_message)
 
         assert isinstance(event, message_events.DMMessageDeleteEvent)
         assert event.app is mock_app
         assert event.shard is mock_shard
+        assert event.old_message is old_message
         assert event.channel_id == 541123
-        assert event.message_ids == {5412}
-        assert event.is_bulk is False
+        assert event.message_id == 5412
 
-    def test_deserialize_message_delete_bulk_event_in_guild(self, event_factory, mock_app, mock_shard):
+    def test_deserialize_guild_message_delete_bulk_event(self, event_factory, mock_app, mock_shard):
         mock_payload = {"ids": ["6523423", "345123"], "channel_id": "564123", "guild_id": "4394949"}
+        old_messages = object()
 
-        event = event_factory.deserialize_message_delete_bulk_event(mock_shard, mock_payload)
+        event = event_factory.deserialize_guild_message_delete_bulk_event(
+            mock_shard, mock_payload, old_messages=old_messages
+        )
 
-        assert isinstance(event, message_events.GuildMessageDeleteEvent)
+        assert isinstance(event, message_events.GuildBulkMessageDeleteEvent)
         assert event.app is mock_app
         assert event.shard is mock_shard
+        assert event.old_messages is old_messages
         assert event.channel_id == 564123
         assert event.message_ids == {6523423, 345123}
-        assert event.is_bulk is True
         assert event.guild_id == 4394949
 
-    def test_deserialize_message_delete_bulk_event_in_dm(self, event_factory, mock_app, mock_shard):
-        mock_payload = {"ids": ["6523423", "345123"], "channel_id": "564123"}
-
-        event = event_factory.deserialize_message_delete_bulk_event(mock_shard, mock_payload)
-
-        assert isinstance(event, message_events.DMMessageDeleteEvent)
-        assert event.app is mock_app
-        assert event.shard is mock_shard
-        assert event.channel_id == 564123
-        assert event.message_ids == {6523423, 345123}
-        assert event.is_bulk is True
-
     ###################
     # REACTION EVENTS #
     ###################
diff --git a/tests/hikari/impl/test_event_manager.py b/tests/hikari/impl/test_event_manager.py
index 9840fc76..c6b8ea51 100644
--- a/tests/hikari/impl/test_event_manager.py
+++ b/tests/hikari/impl/test_event_manager.py
@@ -926,41 +926,45 @@ async def test_on_message_update_stateless(self, stateless_event_manager, shard,
         )
 
     @pytest.mark.asyncio()
-    async def test_on_message_delete_stateless(self, event_manager, shard, event_factory):
-        payload = {}
-        event = mock.Mock(message_id=123)
-
-        event_factory.deserialize_message_delete_event.return_value = event
+    async def test_on_message_delete_stateful(self, event_manager, shard, event_factory):
+        payload = {"id": 123}
 
         await event_manager.on_message_delete(shard, payload)
 
         event_manager._cache.delete_message.assert_called_once_with(123)
-        event_factory.deserialize_message_delete_event.assert_called_once_with(shard, payload)
-        event_manager.dispatch.assert_awaited_once_with(event)
+        event_factory.deserialize_message_delete_event.assert_called_once_with(
+            shard, payload, old_message=event_manager._cache.delete_message.return_value
+        )
+        event_manager.dispatch.assert_awaited_once_with(event_factory.deserialize_message_delete_event.return_value)
 
     @pytest.mark.asyncio()
-    async def test_on_message_delete_stateful(self, stateless_event_manager, shard, event_factory):
+    async def test_on_message_delete_stateless(self, stateless_event_manager, shard, event_factory):
         payload = {}
 
         await stateless_event_manager.on_message_delete(shard, payload)
 
-        event_factory.deserialize_message_delete_event.assert_called_once_with(shard, payload)
+        event_factory.deserialize_message_delete_event.assert_called_once_with(shard, payload, old_message=None)
         stateless_event_manager.dispatch.assert_awaited_once_with(
             event_factory.deserialize_message_delete_event.return_value
         )
 
     @pytest.mark.asyncio()
     async def test_on_message_delete_bulk_stateful(self, event_manager, shard, event_factory):
-        payload = {}
-        event = mock.Mock(message_ids=[123, 456, 789])
-
-        event_factory.deserialize_message_delete_bulk_event.return_value = event
+        payload = {"ids": [123, 456, 789]}
+        message1 = object()
+        message2 = object()
+        message3 = object()
+        event_manager._cache.delete_message.side_effect = [message1, message2, message3]
 
         await event_manager.on_message_delete_bulk(shard, payload)
 
         event_manager._cache.delete_message.assert_has_calls([mock.call(123), mock.call(456), mock.call(789)])
-        event_factory.deserialize_message_delete_bulk_event.assert_called_once_with(shard, payload)
-        event_manager.dispatch.assert_awaited_once_with(event)
+        event_factory.deserialize_guild_message_delete_bulk_event.assert_called_once_with(
+            shard, payload, old_messages={123: message1, 456: message2, 789: message3}
+        )
+        event_manager.dispatch.assert_awaited_once_with(
+            event_factory.deserialize_guild_message_delete_bulk_event.return_value
+        )
 
     @pytest.mark.asyncio()
     async def test_on_message_delete_bulk_stateless(self, stateless_event_manager, shard, event_factory):
@@ -968,9 +972,11 @@ async def test_on_message_delete_bulk_stateless(self, stateless_event_manager, s
 
         await stateless_event_manager.on_message_delete_bulk(shard, payload)
 
-        event_factory.deserialize_message_delete_bulk_event.assert_called_once_with(shard, payload)
+        event_factory.deserialize_guild_message_delete_bulk_event.assert_called_once_with(
+            shard, payload, old_messages={}
+        )
         stateless_event_manager.dispatch.assert_awaited_once_with(
-            event_factory.deserialize_message_delete_bulk_event.return_value
+            event_factory.deserialize_guild_message_delete_bulk_event.return_value
         )
 
     @pytest.mark.asyncio()
diff --git a/tests/hikari/impl/test_event_manager_base.py b/tests/hikari/impl/test_event_manager_base.py
index 8ff38beb..43e6ce2f 100644
--- a/tests/hikari/impl/test_event_manager_base.py
+++ b/tests/hikari/impl/test_event_manager_base.py
@@ -74,37 +74,45 @@ def mock_app():
 
 
 class TestEventStream:
-    @pytest.mark.asyncio()
-    async def test___aenter___and___aexit__(
+    def test___enter___and___exit__(
         self,
     ):
         stub_stream = hikari_test_helpers.mock_class_namespace(
-            event_manager_base.EventStream, open=mock.AsyncMock(), close=mock.AsyncMock()
+            event_manager_base.EventStream, open=mock.Mock(), close=mock.Mock()
         )(mock_app, base_events.Event, timeout=None)
 
-        async with stub_stream:
-            stub_stream.open.assert_awaited_once()
+        with stub_stream:
+            stub_stream.open.assert_called_once_with()
             stub_stream.close.assert_not_called()
 
-        stub_stream.open.assert_awaited_once()
-        stub_stream.close.assert_awaited_once()
+        stub_stream.open.assert_called_once_with()
+        stub_stream.close.assert_called_once_with()
 
-    def test___enter__(self):
+    @pytest.mark.asyncio()
+    async def test___aenter__(self):
         # flake8 gets annoyed if we use "with" here so here's a hacky alternative
-        stub_stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
-            mock_app, base_events.Event, timeout=None
-        )
+        stub_stream = hikari_test_helpers.mock_class_namespace(
+            event_manager_base.EventStream, open=mock.Mock(), close=mock.Mock()
+        )(mock_app, base_events.Event, timeout=None)
+
+        with warnings.catch_warnings():
+            warnings.filterwarnings("ignore", category=DeprecationWarning)
 
-        with pytest.raises(TypeError, match=" is async-only, did you mean 'async with'?"):
-            stub_stream.__enter__()
+            async with stub_stream:
+                stub_stream.open.assert_called_once_with()
+                stub_stream.close.assert_not_called()
 
-    def test___exit__(self):
+        stub_stream.open.assert_called_once_with()
+        stub_stream.close.assert_called_once_with()
+
+    @pytest.mark.asyncio()
+    async def test___exit__(self):
         stub_stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
             mock_app, base_events.Event, timeout=None
         )
 
         try:
-            stub_stream.__exit__(None, None, None)
+            await stub_stream.__aexit__(None, None, None)
         except AttributeError as exc:
             pytest.fail(exc)
 
@@ -115,34 +123,37 @@ async def test__listener_when_filter_returns_false(self, mock_app):
         mock_event = object()
 
         assert await stream._listener(mock_event) is None
-        assert stream._queue.qsize() == 0
+        assert not stream._queue
 
+    @hikari_test_helpers.timeout()
     @pytest.mark.asyncio()
     async def test__listener_when_filter_passes_and_queue_full(self, mock_app):
         stream = event_manager_base.EventStream(mock_app, base_events.Event, timeout=None, limit=2)
-        stream._queue.put_nowait(object())
-        stream._queue.put_nowait(object())
+        stream._queue.append(object())
+        stream._queue.append(object())
         stream.filter(lambda _: True)
         mock_event = object()
 
-        assert await stream._listener(mock_event) is None
-        assert stream._queue.qsize() == 2
-        assert stream._queue.get_nowait() is not mock_event
-        assert stream._queue.get_nowait() is not mock_event
+        with stream:
+            assert await stream._listener(mock_event) is None
+            assert await stream.next() is not mock_event
+            assert await stream.next() is not mock_event
+            assert not stream._queue
 
+    @hikari_test_helpers.timeout()
     @pytest.mark.asyncio()
     async def test__listener_when_filter_passes_and_queue_not_full(self, mock_app):
         stream = event_manager_base.EventStream(mock_app, base_events.Event, timeout=None, limit=None)
-        stream._queue.put_nowait(object())
-        stream._queue.put_nowait(object())
+        stream._queue.append(object())
+        stream._queue.append(object())
         stream.filter(lambda _: True)
         mock_event = object()
 
-        assert await stream._listener(mock_event) is None
-        assert stream._queue.qsize() == 3
-        assert stream._queue.get_nowait() is not mock_event
-        assert stream._queue.get_nowait() is not mock_event
-        assert stream._queue.get_nowait() is mock_event
+        with stream:
+            assert await stream._listener(mock_event) is None
+            assert await stream.next() is not mock_event
+            assert await stream.next() is not mock_event
+            assert await stream.next() is mock_event
 
     @pytest.mark.asyncio()
     @hikari_test_helpers.timeout()
@@ -160,9 +171,9 @@ async def test___anext___times_out(self):
             mock.Mock(), event_type=base_events.Event, timeout=hikari_test_helpers.REASONABLE_QUICK_RESPONSE_TIME
         )
 
-        async with streamer:
-            async for _ in streamer:
-                pytest.fail("streamer shouldn't have yielded anything")
+        with streamer:
+            with pytest.raises(LookupError):
+                await streamer.next()
 
     @pytest.mark.asyncio()
     @hikari_test_helpers.timeout()
@@ -174,16 +185,12 @@ async def test___anext___waits_for_next_event(self):
 
         async def add_event():
             await asyncio.sleep(hikari_test_helpers.REASONABLE_SLEEP_TIME)
-            streamer._queue.put_nowait(mock_event)
+            await streamer._listener(mock_event)
 
         asyncio.create_task(add_event())
 
-        async with streamer:
-            async for event in streamer:
-                assert event is mock_event
-                return
-
-            pytest.fail("streamer should've yielded something")
+        with streamer:
+            assert await streamer.next() is mock_event
 
     @pytest.mark.asyncio()
     @hikari_test_helpers.timeout()
@@ -194,14 +201,10 @@ async def test___anext__(self):
             event_type=base_events.Event,
             timeout=hikari_test_helpers.REASONABLE_QUICK_RESPONSE_TIME,
         )
-        streamer._queue.put_nowait(mock_event)
+        streamer._queue.append(mock_event)
 
-        async with streamer:
-            async for event in streamer:
-                assert event is mock_event
-                return
-
-        pytest.fail("streamer should've yielded something")
+        with streamer:
+            assert await streamer.next() is mock_event
 
     @pytest.mark.asyncio()
     async def test___await__(self):
@@ -210,16 +213,16 @@ async def test___await__(self):
         mock_event_2 = object()
         streamer = hikari_test_helpers.mock_class_namespace(
             event_manager_base.EventStream,
-            close=mock.AsyncMock(),
-            open=mock.AsyncMock(),
+            close=mock.Mock(),
+            open=mock.Mock(),
             init_=False,
             _active=False,
             __anext__=mock.AsyncMock(side_effect=[mock_event_0, mock_event_1, mock_event_2]),
         )()
 
         assert await streamer == [mock_event_0, mock_event_1, mock_event_2]
-        streamer.open.assert_awaited_once()
-        streamer.close.assert_awaited_once()
+        streamer.open.assert_called_once_with()
+        streamer.close.assert_called_once_with()
 
     def test___del___for_active_stream(self):
         mock_coroutine = object()
@@ -230,14 +233,12 @@ def test___del___for_active_stream(self):
         streamer._event_type = base_events.Event
         streamer._active = True
 
-        with mock.patch.object(asyncio, "ensure_future", side_effect=RuntimeError) as ensure_future:
-            with unittest.TestCase().assertLogs("hikari.event_manager", level=logging.WARNING) as logging_watcher:
-                del streamer
+        with unittest.TestCase().assertLogs("hikari.event_manager", level=logging.WARNING) as logging_watcher:
+            del streamer
 
         assert logging_watcher.output == [
             "WARNING:hikari.event_manager:active 'Event' streamer fell out of scope before being closed"
         ]
-        ensure_future.assert_called_once_with(mock_coroutine)
         close_method.assert_called_once_with()
 
     def test___del___for_inactive_stream(self):
@@ -248,37 +249,29 @@ def test___del___for_inactive_stream(self):
         streamer._event_type = base_events.Event
         streamer._active = False
 
-        with mock.patch.object(asyncio, "ensure_future"):
-            del streamer
-            asyncio.ensure_future.assert_not_called()
-
+        del streamer
         close_method.assert_not_called()
 
-    @pytest.mark.asyncio()
-    async def test_close_for_inactive_stream(self, mock_app):
+    def test_close_for_inactive_stream(self, mock_app):
         stream = event_manager_base.EventStream(mock_app, base_events.Event, timeout=None, limit=None)
-        await stream.close()
+        stream.close()
         mock_app.event_manager.unsubscribe.assert_not_called()
 
-    @pytest.mark.asyncio()
-    @hikari_test_helpers.timeout()
-    async def test_close_for_active_stream(self):
+    def test_close_for_active_stream(self):
         mock_registered_listener = object()
         mock_manager = mock.Mock()
         stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
             event_manager=mock_manager, event_type=base_events.Event, timeout=float("inf")
         )
 
-        await stream.open()
+        stream.open()
         stream._registered_listener = mock_registered_listener
-        await stream.close()
+        stream.close()
         mock_manager.unsubscribe.assert_called_once_with(base_events.Event, mock_registered_listener)
         assert stream._active is False
         assert stream._registered_listener is None
 
-    @pytest.mark.asyncio()
-    @hikari_test_helpers.timeout()
-    async def test_close_for_active_stream_handles_value_error(self):
+    def test_close_for_active_stream_handles_value_error(self):
         mock_registered_listener = object()
         mock_manager = mock.Mock()
         mock_manager.unsubscribe.side_effect = ValueError
@@ -286,14 +279,15 @@ async def test_close_for_active_stream_handles_value_error(self):
             event_manager=mock_manager, event_type=base_events.Event, timeout=float("inf")
         )
 
-        await stream.open()
+        stream.open()
         stream._registered_listener = mock_registered_listener
-        await stream.close()
+        stream.close()
         mock_manager.unsubscribe.assert_called_once_with(base_events.Event, mock_registered_listener)
         assert stream._active is False
         assert stream._registered_listener is None
 
-    def test_filter_for_inactive_stream(self):
+    @pytest.mark.asyncio()
+    async def test_filter(self):
         stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
             event_manager=mock.Mock(), event_type=base_events.Event, timeout=1
         )
@@ -308,30 +302,37 @@ def predicate(obj):
 
         stream.filter(predicate, attr=True)
 
-        assert stream._filters(first_pass) is True
-        assert stream._filters(first_fails) is False
-        assert stream._filters(second_pass) is True
-        assert stream._filters(second_fail) is False
+        await stream._listener(first_pass)
+        await stream._listener(first_fails)
+        await stream._listener(second_pass)
+        await stream._listener(second_fail)
+
+        assert await stream == [first_pass, second_pass]
 
     @pytest.mark.asyncio()
-    async def test_filter_for_active_stream(self):
+    async def test_filter_handles_calls_while_active(self):
         stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
-            event_manager=mock.Mock(), event_type=base_events.Event, timeout=float("inf")
+            event_manager=mock.Mock(), event_type=base_events.Event, timeout=1
         )
-        stream._active = True
-        mock_wrapping_iterator = object()
-        predicate = object()
+        stream._filters = iterators.All(())
+        first_pass = mock.Mock(attr=True)
+        second_pass = mock.Mock(attr=True)
+        first_fails = mock.Mock(attr=True)
+        second_fail = mock.Mock(attr=False)
+        await stream._listener(first_pass)
+        await stream._listener(first_fails)
+        await stream._listener(second_pass)
+        await stream._listener(second_fail)
 
-        with mock.patch.object(iterators.LazyIterator, "filter", return_value=mock_wrapping_iterator):
-            assert stream.filter(predicate, name="OK") is mock_wrapping_iterator
+        def predicate(obj):
+            return obj in (first_pass, second_pass)
 
-            iterators.LazyIterator.filter.assert_called_once_with(predicate, name="OK")
+        with stream:
+            stream.filter(predicate, attr=True)
 
-        # Ensure we don't get a warning or error on del
-        stream._active = False
+            assert await stream == [first_pass, second_pass]
 
-    @pytest.mark.asyncio()
-    async def test_open_for_inactive_stream(self):
+    def test_open_for_inactive_stream(self):
         mock_listener = object()
         mock_manager = mock.Mock()
         stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
@@ -345,7 +346,7 @@ async def test_open_for_inactive_stream(self):
 
         with mock.patch.object(event_manager_base, "_generate_weak_listener"):
             with mock.patch.object(weakref, "WeakMethod"):
-                await stream.open()
+                stream.open()
 
                 weakref.WeakMethod.assert_not_called()
             event_manager_base._generate_weak_listener.assert_not_called()
@@ -357,9 +358,7 @@ async def test_open_for_inactive_stream(self):
         # Ensure we don't get a warning or error on del
         stream._active = False
 
-    @pytest.mark.asyncio()
-    @hikari_test_helpers.timeout()
-    async def test_open_for_active_stream(self):
+    def test_open_for_active_stream(self):
         mock_manager = mock.Mock()
         stream = hikari_test_helpers.mock_class_namespace(event_manager_base.EventStream)(
             event_manager=mock_manager, event_type=base_events.Event, timeout=float("inf")
@@ -370,7 +369,7 @@ async def test_open_for_active_stream(self):
 
         with mock.patch.object(event_manager_base, "_generate_weak_listener", return_value=mock_listener):
             with mock.patch.object(weakref, "WeakMethod", return_value=mock_listener_ref):
-                await stream.open()
+                stream.open()
 
                 weakref.WeakMethod.assert_called_once_with(stream._listener)
             event_manager_base._generate_weak_listener.assert_called_once_with(mock_listener_ref)