# Path: il2fb/ds/airbridge/json.py
# class JSONEncoder(_json.JSONEncoder):
#     def default(self, obj):
# def object_decoder(obj):
# 
# Path: il2fb/ds/airbridge/streaming/subscribers/base.py
# class PluggableStreamingSubscriber(StreamingSubscriber):
# 
#     def __init__(self, app):
#         self._app = app
# 
#     def plug_in(self) -> None:
#         pass
# 
#     async def wait_plugged(self) -> Awaitable[None]:
#         pass
# 
#     def unplug(self) -> None:
#         pass
# 
#     async def wait_unplugged(self) -> Awaitable[None]:
#         pass


import asyncio
import logging
from typing import Any, Awaitable
from nats_stream.aio.publisher import Publisher
from il2fb.ds.airbridge import json
from il2fb.ds.airbridge.streaming.subscribers.base import PluggableStreamingSubscriber
        self._publisher = Publisher(
            sc=self._client,
            subject=self._subject,
            ack_cb=self._handle_ack,
        )

        queue_coroutine = self._process_queue()
        self._queue_task = self._app.loop.create_task(queue_coroutine)

    def _handle_ack(self, ack) -> None:
        LOG.debug(f"nats streaming ack (guid={ack}, subject={self._subject})")

    def unplug(self) -> None:
        if self._queue_task:
            self._queue.put_nowait(None)

    async def wait_unplugged(self) -> Awaitable[None]:
        if self._queue_task:
            await self._queue_task

    async def write(self, o: Any) -> Awaitable[None]:
        msg = json.dumps(o).encode()
        await self._queue.put(msg)

    async def _process_queue(self) -> Awaitable[None]:
        LOG.info(
            f"processing of nats streaming queue was started "
            f"(subject={self._subject})"
        )


# Path: il2fb/ds/airbridge/json.py
# class JSONEncoder(_json.JSONEncoder):
#     def default(self, obj):
# def object_decoder(obj):
# 
# Path: il2fb/ds/airbridge/streaming/subscribers/base.py
# class PluggableStreamingSubscriber(StreamingSubscriber):
# 
#     def __init__(self, app):
#         self._app = app
# 
#     def plug_in(self) -> None:
#         pass
# 
#     async def wait_plugged(self) -> Awaitable[None]:
#         pass
# 
#     def unplug(self) -> None:
#         pass
# 
#     async def wait_unplugged(self) -> Awaitable[None]:
#         pass


# Path: il2fb/ds/airbridge/streaming/subscribers/nats.py
import asyncio
import logging
from typing import Any, Awaitable
from nats_stream.aio.publisher import Publisher
from il2fb.ds.airbridge import json
from il2fb.ds.airbridge.streaming.subscribers.base import PluggableStreamingSubscriber

        self._publisher = Publisher(
            sc=self._client,
            subject=self._subject,
            ack_cb=self._handle_ack,
        )

        queue_coroutine = self._process_queue()
        self._queue_task = self._app.loop.create_task(queue_coroutine)

    def _handle_ack(self, ack) -> None:
        LOG.debug(f"nats streaming ack (guid={ack}, subject={self._subject})")

    def unplug(self) -> None:
        if self._queue_task:
            self._queue.put_nowait(None)

    async def wait_unplugged(self) -> Awaitable[None]:
        if self._queue_task:
            await self._queue_task

    async def write(self, o: Any) -> Awaitable[None]:
        msg = json.dumps(o).encode()
        await self._queue.put(msg)

    async def _process_queue(self) -> Awaitable[None]:
        LOG.info(
            f"processing of nats streaming queue was started "
            f"(subject={self._subject})"
        )


        while True: