class CustomHttpStream(HttpStream):
    def __init__(self, client: ApiClient, ref: ConversationReference):
        super().__init__(client, ref)
        self._is_canceled = False

    def emit(self, activity: Union[MessageActivityInput, TypingActivityInput, str]) -> None:
        if self.is_canceled():
            logger.warning("Not emitting activity, stream is canceled.")
            return None
        return super().emit(activity)

    async def _send(self, to_send: Union[TypingActivityInput, MessageActivityInput]) -> SentActivity:
        try:
            res = await super()._send(to_send)
            return res
        except HTTPStatusError as e:
            if e.response.status_code == 403:
                # Set canceled and return a `CancelledError` so that the
                # request won't be retried.
                self._is_canceled = True
                logger.warning("Teams channel stopped the stream.")
                raise asyncio.CancelledError("Teams channel stopped the stream.")
            raise e
        except Exception as e:
            raise e

    def is_canceled(self) -> bool:
        return self._is_canceled