diff --git a/faust/streams.py b/faust/streams.py
index a93b722d5..8bc31a61d 100644
--- a/faust/streams.py
+++ b/faust/streams.py
@@ -682,7 +682,7 @@ def enumerate(self, start: int = 0) -> AsyncIterable[Tuple[int, T_co]]:
     @_tracks_buffer_agen
     async def noack_take(
         self, max_: int, within: Seconds
-    ) -> AsyncGenerator[Sequence[T_co], None]:
+    ) -> AsyncIterable[Sequence[EventT[T_co], None]]:
         """
          Buffer n values at a time and yield a list of buffered values.
         :param max_: Max number of messages to receive. When more than this
@@ -694,7 +694,7 @@ async def noack_take(
              the agent is likely to stall and block buffered events for an
              unreasonable length of time(!).
         """
-        buffer: List[T_co] = []
+        buffer: List[EventT[T_co]] = []
         events: List[EventT] = []
         buffer_add = buffer.append
         event_add = events.append
@@ -723,7 +723,7 @@ async def add_to_buffer(value: T) -> T:
 
                 # We want to save events instead of values to allow for manual ack
                 event = self.current_event
-                buffer_add(cast(T_co, event))
+                buffer_add(event)
                 if event is None:
                     raise RuntimeError("Take buffer found current_event is None")
 
diff --git a/faust/types/events.py b/faust/types/events.py
index b1deaa804..5831d7d8d 100644
--- a/faust/types/events.py
+++ b/faust/types/events.py
@@ -34,7 +34,7 @@ class _SchemaT: ...  # noqa
 class EventT(Generic[T], AsyncContextManager):
     app: _AppT
     key: K
-    value: V
+    value: T
     headers: Mapping
     message: Message
     acked: bool
diff --git a/faust/types/streams.py b/faust/types/streams.py
index b81159727..fb0cc8aee 100644
--- a/faust/types/streams.py
+++ b/faust/types/streams.py
@@ -16,7 +16,6 @@
     Tuple,
     TypeVar,
     Union,
-    no_type_check,
 )
 
 from mode import Seconds, ServiceT
@@ -149,22 +148,35 @@ def info(self) -> Mapping[str, Any]: ...
     def clone(self, **kwargs: Any) -> "StreamT": ...
 
     @abc.abstractmethod
-    @no_type_check
-    async def items(self) -> AsyncIterator[Tuple[K, T_co]]: ...
+    def noack(self) -> "StreamT": ...
 
     @abc.abstractmethod
-    @no_type_check
-    async def events(self) -> AsyncIterable[EventT]: ...
+    def items(self) -> AsyncIterator[Tuple[K, T_co]]: ...
 
     @abc.abstractmethod
-    @no_type_check
-    async def take(
+    def events(self) -> AsyncIterable[EventT]: ...
+
+    @abc.abstractmethod
+    def take(self, max_: int, within: Seconds) -> AsyncIterable[Sequence[T_co]]: ...
+
+    @abc.abstractmethod
+    def take_events(
         self, max_: int, within: Seconds
+    ) -> AsyncIterable[Sequence[EventT[T_co]]]: ...
+
+    @abc.abstractmethod
+    def take_with_timestamp(
+        self, max_: int, within: Seconds, timestamp_field_name: str
     ) -> AsyncIterable[Sequence[T_co]]: ...
 
     @abc.abstractmethod
     def enumerate(self, start: int = 0) -> AsyncIterable[Tuple[int, T_co]]: ...
 
+    @abc.abstractmethod
+    def noack_take(
+        self, max_: int, within: Seconds
+    ) -> AsyncIterable[Sequence[EventT[T_co]]]: ...
+
     @abc.abstractmethod
     def through(self, channel: Union[str, ChannelT]) -> "StreamT": ...
 
@@ -180,6 +192,9 @@ def group_by(
         topic: Optional[TopicT] = None,
     ) -> "StreamT": ...
 
+    @abc.abstractmethod
+    def filter(self, fun: Processor[T]) -> "StreamT": ...
+
     @abc.abstractmethod
     def derive_topic(
         self,
@@ -204,8 +219,5 @@ def __iter__(self) -> Any: ...
     @abc.abstractmethod
     def __next__(self) -> T_co: ...
 
-    @abc.abstractmethod
-    def __aiter__(self) -> AsyncIterator[T_co]: ...
-
     @abc.abstractmethod
     async def ack(self, event: EventT) -> bool: ...
diff --git a/faust/types/transports.py b/faust/types/transports.py
index 12ad90d0d..668d71164 100644
--- a/faust/types/transports.py
+++ b/faust/types/transports.py
@@ -59,7 +59,7 @@ class _AppT: ...  # noqa
 
 #: Argument to Consumer.commit to specify topics/tps to commit.
 TPorTopic = Union[str, TP]
-TPorTopicSet = AbstractSet[TPorTopic]
+TPorTopicSet = Optional[AbstractSet[TPorTopic]]
 
 #: Callback (:keyword:`async def`) called when consumer partitions are revoked.
 PartitionsRevokedCallback = Callable[[Set[TP]], Awaitable[None]]