Centralize event stream wrappers

Author: JordonPhillips

packages/aws-event-stream/src/aws_event_stream/_private/deserializers.py

@@ -1,20 +1,16 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
-import asyncio
 import datetime
 from collections.abc import Callable
 
-from smithy_core.aio.interfaces import AsyncByteStream
 from smithy_core.codecs import Codec
 from smithy_core.deserializers import (
-    DeserializeableShape,
     ShapeDeserializer,
     SpecificShapeDeserializer,
 )
 from smithy_core.schemas import Schema
 from smithy_core.shapes import ShapeType
 from smithy_core.utils import expect_type
-from smithy_core.aio.interfaces.eventstream import AsyncEventReceiver
 
 from ..events import HEADERS_DICT, Event
 from ..exceptions import EventError, UnmodeledEventError
@@ -28,40 +24,6 @@
 INITIAL_MESSAGE_TYPES = (INITIAL_REQUEST_EVENT_TYPE, INITIAL_RESPONSE_EVENT_TYPE)
 
 
-class AWSAsyncEventReceiver[E: DeserializeableShape](AsyncEventReceiver[E]):
-    def __init__(
-        self,
-        payload_codec: Codec,
-        source: AsyncByteStream,
-        deserializer: Callable[[ShapeDeserializer], E],
-        is_client_mode: bool = True,
-    ) -> None:
-        self._payload_codec = payload_codec
-        self._source = source
-        self._is_client_mode = is_client_mode
-        self._deserializer = deserializer
-
-    async def receive(self) -> E | None:
-        event = await Event.decode_async(self._source)
-        if event is None:
-            return None
-
-        deserializer = EventDeserializer(
-            event=event,
-            payload_codec=self._payload_codec,
-            is_client_mode=self._is_client_mode,
-        )
-        result = self._deserializer(deserializer)
-        if isinstance(getattr(result, "value"), Exception):
-            raise result.value  # type: ignore
-        return result
-
-    async def close(self) -> None:
-        if (close := getattr(self._source, "close", None)) is not None:
-            if asyncio.iscoroutine(result := close()):
-                await result
-
-
 class EventDeserializer(SpecificShapeDeserializer):
     def __init__(
         self, event: Event, payload_codec: Codec, is_client_mode: bool = True

packages/aws-event-stream/src/aws_event_stream/_private/serializers.py

@@ -1,24 +1,19 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
-import asyncio
 import datetime
-from collections.abc import Callable, Iterator
+from collections.abc import Iterator
 from contextlib import contextmanager
 from io import BytesIO
 from typing import Never
 
-from smithy_core.aio.interfaces import AsyncWriter
 from smithy_core.codecs import Codec
-from smithy_core.exceptions import ExpectationNotMetException
 from smithy_core.schemas import Schema
 from smithy_core.serializers import (
     InterceptingSerializer,
-    SerializeableShape,
     ShapeSerializer,
     SpecificShapeSerializer,
 )
 from smithy_core.shapes import ShapeType
-from smithy_core.aio.interfaces.eventstream import AsyncEventPublisher
 
 from ..events import EventMessage, HEADER_VALUE, Short, Byte, Long
 from ..exceptions import InvalidHeaderValue
@@ -33,41 +28,6 @@
 _DEFAULT_BLOB_CONTENT_TYPE = "application/octet-stream"
 
 
-type Signer = Callable[[EventMessage], EventMessage]
-"""A function that takes an event message and signs it, and returns it signed."""
-
-
-class AWSAsyncEventPublisher[E: SerializeableShape](AsyncEventPublisher[E]):
-    def __init__(
-        self,
-        payload_codec: Codec,
-        async_writer: AsyncWriter,
-        signer: Signer | None = None,
-        is_client_mode: bool = True,
-    ):
-        self._writer = async_writer
-        self._signer = signer
-        self._serializer = EventSerializer(
-            payload_codec=payload_codec, is_client_mode=is_client_mode
-        )
-
-    async def send(self, event: E) -> None:
-        event.serialize(self._serializer)
-        result = self._serializer.get_result()
-        if result is None:
-            raise ExpectationNotMetException(
-                "Expected an event message to be serialized, but was None."
-            )
-        if self._signer is not None:
-            result = self._signer(result)
-        await self._writer.write(result.encode())
-
-    async def close(self) -> None:
-        if (close := getattr(self._writer, "close", None)) is not None:
-            if asyncio.iscoroutine(result := close()):
-                await result
-
-
 class EventSerializer(SpecificShapeSerializer):
     def __init__(
         self,