# Path: il2fb/ds/airbridge/json.py
# class JSONEncoder(_json.JSONEncoder):
#     def default(self, obj):
# def object_decoder(obj):
# 
# Path: il2fb/ds/airbridge/api/http/responses/ws.py
# def WSSuccess(payload=None, detail=None):
#     payload = payload or {}
#     payload['status'] = WS_STATUS.SUCCESS
# 
#     if detail:
#         payload['detail'] = str(detail)
# 
#     return json.dumps(payload)
# 
# def WSFailure(payload=None, detail=None):
#     payload = payload or {}
#     payload.update({
#         'status': WS_STATUS.FAILURE,
#         'detail': str(detail or "Request execution error."),
#     })
#     return json.dumps(payload)
# 
# Path: il2fb/ds/airbridge/api/http/security.py
# def with_authorization(view):
# 
#     @functools.wraps(view)
#     async def wrapper(arg0, *args, **kwargs):
#         request = arg0 if isinstance(arg0, web.Request) else arg0.request
#         backend = request.app.get('auth_backend')
# 
#         if backend and not backend.authorize(request):
#             raise web.HTTPForbidden()
# 
#         return (await view(arg0, *args, **kwargs))
# 
#     return wrapper
# 
# Path: il2fb/ds/airbridge/streaming/subscribers/base.py
# class StreamingSubscriber(metaclass=abc.ABCMeta):
# 
#     @abc.abstractmethod
#     async def write(self, o: Any) -> Awaitable[None]:
#         pass


import logging
from enum import IntEnum
from typing import Any, Awaitable
from aiohttp import web, WSMsgType
from il2fb.ds.airbridge import json
from il2fb.ds.airbridge.api.http.responses.ws import WSSuccess, WSFailure
from il2fb.ds.airbridge.api.http.security import with_authorization
from il2fb.ds.airbridge.streaming.subscribers.base import StreamingSubscriber
        async for msg in self._ws:
            if msg.type == WSMsgType.TEXT:
                await self._on_message(msg.data)
            elif msg.type == WSMsgType.ERROR:
                e = self._ws.exception()
                LOG.error(
                    f"ws streaming connection was closed unexpectedly: {e}"
                )

        await self._unsubscribe_from_all()

        LOG.debug("ws streaming connection was closed")
        return self._ws

    async def _on_message(self, message: str) -> Awaitable[None]:
        LOG.debug(f"ws streaming message {repr(message)}")

        if message == "close":
            await self._unsubscribe_from_all()
            await self._ws.close()
        else:
            try:
                data = json.loads(message)
                result = await self._on_data(data)
            except Exception as e:
                LOG.exception(
                    f"failed to handle ws streaming message {repr(message)}"
                )
                await self._ws.send_str(WSFailure(detail=e))
            else:

# Path: il2fb/ds/airbridge/json.py
# class JSONEncoder(_json.JSONEncoder):
#     def default(self, obj):
# def object_decoder(obj):
# 
# Path: il2fb/ds/airbridge/api/http/responses/ws.py
# def WSSuccess(payload=None, detail=None):
#     payload = payload or {}
#     payload['status'] = WS_STATUS.SUCCESS
# 
#     if detail:
#         payload['detail'] = str(detail)
# 
#     return json.dumps(payload)
# 
# def WSFailure(payload=None, detail=None):
#     payload = payload or {}
#     payload.update({
#         'status': WS_STATUS.FAILURE,
#         'detail': str(detail or "Request execution error."),
#     })
#     return json.dumps(payload)
# 
# Path: il2fb/ds/airbridge/api/http/security.py
# def with_authorization(view):
# 
#     @functools.wraps(view)
#     async def wrapper(arg0, *args, **kwargs):
#         request = arg0 if isinstance(arg0, web.Request) else arg0.request
#         backend = request.app.get('auth_backend')
# 
#         if backend and not backend.authorize(request):
#             raise web.HTTPForbidden()
# 
#         return (await view(arg0, *args, **kwargs))
# 
#     return wrapper
# 
# Path: il2fb/ds/airbridge/streaming/subscribers/base.py
# class StreamingSubscriber(metaclass=abc.ABCMeta):
# 
#     @abc.abstractmethod
#     async def write(self, o: Any) -> Awaitable[None]:
#         pass


# Path: il2fb/ds/airbridge/api/http/views/streaming.py
import logging
from enum import IntEnum
from typing import Any, Awaitable
from aiohttp import web, WSMsgType
from il2fb.ds.airbridge import json
from il2fb.ds.airbridge.api.http.responses.ws import WSSuccess, WSFailure
from il2fb.ds.airbridge.api.http.security import with_authorization
from il2fb.ds.airbridge.streaming.subscribers.base import StreamingSubscriber

        async for msg in self._ws:
            if msg.type == WSMsgType.TEXT:
                await self._on_message(msg.data)
            elif msg.type == WSMsgType.ERROR:
                e = self._ws.exception()
                LOG.error(
                    f"ws streaming connection was closed unexpectedly: {e}"
                )

        await self._unsubscribe_from_all()

        LOG.debug("ws streaming connection was closed")
        return self._ws

    async def _on_message(self, message: str) -> Awaitable[None]:
        LOG.debug(f"ws streaming message {repr(message)}")

        if message == "close":
            await self._unsubscribe_from_all()
            await self._ws.close()
        else:
            try:
                data = json.loads(message)
                result = await self._on_data(data)
            except Exception as e:
                LOG.exception(
                    f"failed to handle ws streaming message {repr(message)}"
                )
                await self._ws.send_str(WSFailure(detail=e))
            else:

                await self._ws.send_str(WSSuccess(payload=result))