# Path: gql/transport/async_transport.py
# class AsyncTransport:
#     @abc.abstractmethod
#     async def connect(self):
#         """Coroutine used to create a connection to the specified address
#         """
#         raise NotImplementedError(
#             "Any AsyncTransport subclass must implement connect method"
#         )  # pragma: no cover
# 
#     @abc.abstractmethod
#     async def close(self):
#         """Coroutine used to Close an established connection
#         """
#         raise NotImplementedError(
#             "Any AsyncTransport subclass must implement close method"
#         )  # pragma: no cover
# 
#     @abc.abstractmethod
#     async def execute(
#         self,
#         document: DocumentNode,
#         variable_values: Optional[Dict[str, Any]] = None,
#         operation_name: Optional[str] = None,
#     ) -> ExecutionResult:
#         """Execute the provided document AST for either a remote or local GraphQL Schema.
#         """
#         raise NotImplementedError(
#             "Any AsyncTransport subclass must implement execute method"
#         )  # pragma: no cover
# 
#     @abc.abstractmethod
#     def subscribe(
#         self,
#         document: DocumentNode,
#         variable_values: Optional[Dict[str, Any]] = None,
#         operation_name: Optional[str] = None,
#     ) -> AsyncGenerator[ExecutionResult, None]:
#         """Send a query and receive the results using an async generator
# 
#         The query can be a graphql query, mutation or subscription
# 
#         The results are sent as an ExecutionResult object
#         """
#         raise NotImplementedError(
#             "Any AsyncTransport subclass must implement subscribe method"
#         )  # pragma: no cover
# 
# Path: gql/transport/exceptions.py
# class TransportAlreadyConnected(TransportError):
#     """Transport is already connected.
# 
#     Exception generated when the client is trying to connect to the transport
#     while the transport is already connected.
#     """
# 
# class TransportClosed(TransportError):
#     """Transport is already closed.
# 
#     This exception is generated when the client is trying to use the transport
#     while the transport was previously closed.
#     """
# 
# class TransportProtocolError(TransportError):
#     """Transport protocol error.
# 
#     The answer received from the server does not correspond to the transport protocol.
#     """
# 
# class TransportQueryError(Exception):
#     """The server returned an error for a specific query.
# 
#     This exception should not close the transport connection.
#     """
# 
#     query_id: Optional[int]
#     errors: Optional[List[Any]]
#     data: Optional[Any]
#     extensions: Optional[Any]
# 
#     def __init__(
#         self,
#         msg: str,
#         query_id: Optional[int] = None,
#         errors: Optional[List[Any]] = None,
#         data: Optional[Any] = None,
#         extensions: Optional[Any] = None,
#     ):
#         super().__init__(msg)
#         self.query_id = query_id
#         self.errors = errors
#         self.data = data
#         self.extensions = extensions
# 
# class TransportServerError(TransportError):
#     """The server returned a global error.
# 
#     This exception will close the transport connection.
#     """
# 
#     code: Optional[int]
# 
#     def __init__(self, message: str, code: Optional[int] = None):
#         super(TransportServerError, self).__init__(message)
#         self.code = code


import asyncio
import logging
import warnings
import websockets
from abc import abstractmethod
from contextlib import suppress
from ssl import SSLContext
from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple, Union, cast
from graphql import DocumentNode, ExecutionResult
from websockets.client import WebSocketClientProtocol
from websockets.datastructures import HeadersLike
from websockets.exceptions import ConnectionClosed
from websockets.typing import Data, Subprotocol
from .async_transport import AsyncTransport
from .exceptions import (
    TransportAlreadyConnected,
    TransportClosed,
    TransportProtocolError,
    TransportQueryError,
    TransportServerError,
)

    async def _receive_data_loop(self) -> None:
        """Main asyncio task which will listen to the incoming messages and will
        call the parse_answer and handle_answer methods of the subclass."""
        try:
            while True:

                # Wait the next answer from the websocket server
                try:
                    answer = await self._receive()
                except (ConnectionClosed, TransportProtocolError) as e:
                    await self._fail(e, clean_close=False)
                    break
                except TransportClosed:
                    break

                # Parse the answer
                try:
                    answer_type, answer_id, execution_result = self._parse_answer(
                        answer
                    )
                except TransportQueryError as e:
                    # Received an exception for a specific query
                    # ==> Add an exception to this query queue
                    # The exception is raised for this specific query,
                    # but the transport is not closed.
                    assert isinstance(
                        e.query_id, int
                    ), "TransportQueryError should have a query_id defined here"
                    try:

# Path: gql/transport/async_transport.py
# class AsyncTransport:
#     @abc.abstractmethod
#     async def connect(self):
#         """Coroutine used to create a connection to the specified address
#         """
#         raise NotImplementedError(
#             "Any AsyncTransport subclass must implement connect method"
#         )  # pragma: no cover
# 
#     @abc.abstractmethod
#     async def close(self):
#         """Coroutine used to Close an established connection
#         """
#         raise NotImplementedError(
#             "Any AsyncTransport subclass must implement close method"
#         )  # pragma: no cover
# 
#     @abc.abstractmethod
#     async def execute(
#         self,
#         document: DocumentNode,
#         variable_values: Optional[Dict[str, Any]] = None,
#         operation_name: Optional[str] = None,
#     ) -> ExecutionResult:
#         """Execute the provided document AST for either a remote or local GraphQL Schema.
#         """
#         raise NotImplementedError(
#             "Any AsyncTransport subclass must implement execute method"
#         )  # pragma: no cover
# 
#     @abc.abstractmethod
#     def subscribe(
#         self,
#         document: DocumentNode,
#         variable_values: Optional[Dict[str, Any]] = None,
#         operation_name: Optional[str] = None,
#     ) -> AsyncGenerator[ExecutionResult, None]:
#         """Send a query and receive the results using an async generator
# 
#         The query can be a graphql query, mutation or subscription
# 
#         The results are sent as an ExecutionResult object
#         """
#         raise NotImplementedError(
#             "Any AsyncTransport subclass must implement subscribe method"
#         )  # pragma: no cover
# 
# Path: gql/transport/exceptions.py
# class TransportAlreadyConnected(TransportError):
#     """Transport is already connected.
# 
#     Exception generated when the client is trying to connect to the transport
#     while the transport is already connected.
#     """
# 
# class TransportClosed(TransportError):
#     """Transport is already closed.
# 
#     This exception is generated when the client is trying to use the transport
#     while the transport was previously closed.
#     """
# 
# class TransportProtocolError(TransportError):
#     """Transport protocol error.
# 
#     The answer received from the server does not correspond to the transport protocol.
#     """
# 
# class TransportQueryError(Exception):
#     """The server returned an error for a specific query.
# 
#     This exception should not close the transport connection.
#     """
# 
#     query_id: Optional[int]
#     errors: Optional[List[Any]]
#     data: Optional[Any]
#     extensions: Optional[Any]
# 
#     def __init__(
#         self,
#         msg: str,
#         query_id: Optional[int] = None,
#         errors: Optional[List[Any]] = None,
#         data: Optional[Any] = None,
#         extensions: Optional[Any] = None,
#     ):
#         super().__init__(msg)
#         self.query_id = query_id
#         self.errors = errors
#         self.data = data
#         self.extensions = extensions
# 
# class TransportServerError(TransportError):
#     """The server returned a global error.
# 
#     This exception will close the transport connection.
#     """
# 
#     code: Optional[int]
# 
#     def __init__(self, message: str, code: Optional[int] = None):
#         super(TransportServerError, self).__init__(message)
#         self.code = code


# Path: gql/transport/websockets_base.py
import asyncio
import logging
import warnings
import websockets
from abc import abstractmethod
from contextlib import suppress
from ssl import SSLContext
from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple, Union, cast
from graphql import DocumentNode, ExecutionResult
from websockets.client import WebSocketClientProtocol
from websockets.datastructures import HeadersLike
from websockets.exceptions import ConnectionClosed
from websockets.typing import Data, Subprotocol
from .async_transport import AsyncTransport
from .exceptions import (
    TransportAlreadyConnected,
    TransportClosed,
    TransportProtocolError,
    TransportQueryError,
    TransportServerError,
)


    async def _receive_data_loop(self) -> None:
        """Main asyncio task which will listen to the incoming messages and will
        call the parse_answer and handle_answer methods of the subclass."""
        try:
            while True:

                # Wait the next answer from the websocket server
                try:
                    answer = await self._receive()
                except (ConnectionClosed, TransportProtocolError) as e:
                    await self._fail(e, clean_close=False)
                    break
                except TransportClosed:
                    break

                # Parse the answer
                try:
                    answer_type, answer_id, execution_result = self._parse_answer(
                        answer
                    )
                except TransportQueryError as e:
                    # Received an exception for a specific query
                    # ==> Add an exception to this query queue
                    # The exception is raised for this specific query,
                    # but the transport is not closed.
                    assert isinstance(
                        e.query_id, int
                    ), "TransportQueryError should have a query_id defined here"
                    try:

                        await self.listeners[e.query_id].set_exception(e)