"""
This type stub file was generated by pyright.
"""

from typing import Awaitable, Callable, Dict, List, Optional, TYPE_CHECKING
from nats.aio.msg import Msg
from nats.aio.subscription import Subscription
from nats.js import api
from nats.js.kv import KeyValue
from nats.js.manager import JetStreamManager
from nats.js.object_store import ObjectStore
from nats import NATS

if TYPE_CHECKING:
    ...
NATS_HDR_LINE = ...
NATS_HDR_LINE_SIZE = ...
_CRLF_ = ...
_CRLF_LEN_ = ...
KV_STREAM_TEMPLATE = ...
KV_PRE_TEMPLATE = ...
Callback = Callable[['Msg'], Awaitable[None]]
DEFAULT_JS_SUB_PENDING_MSGS_LIMIT = ...
DEFAULT_JS_SUB_PENDING_BYTES_LIMIT = ...
KV_MAX_HISTORY = ...
class JetStreamContext(JetStreamManager):
    """
    Fully featured context for interacting with JetStream.

    :param conn: NATS Connection
    :param prefix: Default JetStream API Prefix.
    :param domain: Optional domain used by the JetStream API.
    :param timeout: Timeout for all JS API actions.

    ::

        import asyncio
        import nats

        async def main():
            nc = await nats.connect()
            js = nc.jetstream()

            await js.add_stream(name='hello', subjects=['hello'])
            ack = await js.publish('hello', b'Hello JS!')
            print(f'Ack: stream={ack.stream}, sequence={ack.seq}')
            # Ack: stream=hello, sequence=1
            await nc.close()

        if __name__ == '__main__':
            asyncio.run(main())

    """
    def __init__(self, conn: NATS, prefix: str = ..., domain: Optional[str] = ..., timeout: float = ...) -> None:
        ...
    
    async def publish(self, subject: str, payload: bytes = ..., timeout: Optional[float] = ..., stream: Optional[str] = ..., headers: Optional[Dict] = ...) -> api.PubAck:
        """
        publish emits a new message to JetStream.
        """
        ...
    
    async def subscribe(self, subject: str, queue: Optional[str] = ..., cb: Optional[Callback] = ..., durable: Optional[str] = ..., stream: Optional[str] = ..., config: Optional[api.ConsumerConfig] = ..., manual_ack: bool = ..., ordered_consumer: bool = ..., idle_heartbeat: Optional[float] = ..., flow_control: bool = ..., pending_msgs_limit: int = ..., pending_bytes_limit: int = ..., deliver_policy: Optional[api.DeliverPolicy] = ..., headers_only: Optional[bool] = ..., inactive_threshold: Optional[float] = ...) -> PushSubscription:
        """Create consumer if needed and push-subscribe to it.

        1. Check if consumer exists.
        2. Creates consumer if needed.
        3. Calls `subscribe_bind`.

        :param subject: Subject from a stream from JetStream.
        :param queue: Deliver group name from a set a of queue subscribers.
        :param durable: Name of the durable consumer to which the the subscription should be bound.
        :param stream: Name of the stream to which the subscription should be bound. If not set,
          then the client will automatically look it up based on the subject.
        :param manual_ack: Disables auto acking for async subscriptions.
        :param ordered_consumer: Enable ordered consumer mode.
        :param idle_heartbeat: Enable Heartbeats for a consumer to detect failures.
        :param flow_control: Enable Flow Control for a consumer.

        ::

            import asyncio
            import nats

            async def main():
                nc = await nats.connect()
                js = nc.jetstream()

                await js.add_stream(name='hello', subjects=['hello'])
                await js.publish('hello', b'Hello JS!')

                async def cb(msg):
                  print('Received:', msg)

                # Ephemeral Async Subscribe
                await js.subscribe('hello', cb=cb)

                # Durable Async Subscribe
                # NOTE: Only one subscription can be bound to a durable name. It also auto acks by default.
                await js.subscribe('hello', cb=cb, durable='foo')

                # Durable Sync Subscribe
                # NOTE: Sync subscribers do not auto ack.
                await js.subscribe('hello', durable='bar')

                # Queue Async Subscribe
                # NOTE: Here 'workers' becomes deliver_group, durable name and queue name.
                await js.subscribe('hello', 'workers', cb=cb)

            if __name__ == '__main__':
                asyncio.run(main())

        """
        ...
    
    async def subscribe_bind(self, stream: str, config: api.ConsumerConfig, consumer: str, cb: Optional[Callback] = ..., manual_ack: bool = ..., ordered_consumer: bool = ..., pending_msgs_limit: int = ..., pending_bytes_limit: int = ...) -> PushSubscription:
        """Push-subscribe to an existing consumer.
        """
        ...
    
    async def pull_subscribe(self, subject: str, durable: Optional[str] = ..., stream: Optional[str] = ..., config: Optional[api.ConsumerConfig] = ..., pending_msgs_limit: int = ..., pending_bytes_limit: int = ..., inbox_prefix: bytes = ...) -> JetStreamContext.PullSubscription:
        """Create consumer and pull subscription.

        1. Find stream name by subject if `stream` is not passed.
        2. Create consumer with the given `config` if not created.
        3. Call `pull_subscribe_bind`.

        ::

            import asyncio
            import nats

            async def main():
                nc = await nats.connect()
                js = nc.jetstream()

                await js.add_stream(name='mystream', subjects=['foo'])
                await js.publish('foo', b'Hello World!')

                sub = await js.pull_subscribe('foo', stream='mystream')

                msgs = await sub.fetch()
                msg = msgs[0]
                await msg.ack()

                await nc.close()

            if __name__ == '__main__':
                asyncio.run(main())

        """
        ...
    
    async def pull_subscribe_bind(self, consumer: Optional[str] = ..., stream: Optional[str] = ..., inbox_prefix: bytes = ..., pending_msgs_limit: int = ..., pending_bytes_limit: int = ..., name: Optional[str] = ..., durable: Optional[str] = ...) -> JetStreamContext.PullSubscription:
        """
        pull_subscribe returns a `PullSubscription` that can be delivered messages
        from a JetStream pull based consumer by calling `sub.fetch`.

        ::

            import asyncio
            import nats

            async def main():
                nc = await nats.connect()
                js = nc.jetstream()

                await js.add_stream(name='mystream', subjects=['foo'])
                await js.publish('foo', b'Hello World!')

                msgs = await sub.fetch()
                msg = msgs[0]
                await msg.ack()

                await nc.close()

            if __name__ == '__main__':
                asyncio.run(main())

        """
        ...
    
    @classmethod
    def is_status_msg(cls, msg: Optional[Msg]) -> Optional[str]:
        ...
    
    class _JSI:
        def __init__(self, js: JetStreamContext, conn: NATS, stream: str, ordered: Optional[bool], psub: JetStreamContext.PushSubscription, sub: Subscription, ccreq: api.ConsumerConfig) -> None:
            ...
        
        def track_sequences(self, reply: str) -> None:
            ...
        
        def schedule_flow_control_response(self, reply: str) -> None:
            ...
        
        def get_js_delivered(self): # -> int:
            ...
        
        async def activity_check(self): # -> None:
            ...
        
        async def check_flow_control_response(self): # -> None:
            ...
        
        async def check_for_sequence_mismatch(self, msg: Msg) -> Optional[bool]:
            ...
        
        async def reset_ordered_consumer(self, sseq: Optional[int]) -> bool:
            ...
        
        async def recreate_consumer(self) -> None:
            ...
        
    
    
    class PushSubscription(Subscription):
        """
        PushSubscription is a subscription that is delivered messages.
        """
        def __init__(self, js: JetStreamContext, sub: Subscription, stream: str, consumer: str) -> None:
            ...
        
        async def consumer_info(self) -> api.ConsumerInfo:
            """
            consumer_info gets the current info of the consumer from this subscription.
            """
            ...
        
        @property
        def delivered(self) -> int:
            """
            Number of delivered messages to this subscription so far.
            """
            ...
        
        @delivered.setter
        def delivered(self, value): # -> None:
            ...
        
        async def next_msg(self, timeout: Optional[float] = ...) -> Msg:
            """
            :params timeout: Time in seconds to wait for next message before timing out.
            :raises nats.errors.TimeoutError:

            next_msg can be used to retrieve the next message from a stream of messages using
            await syntax, this only works when not passing a callback on `subscribe`::
            """
            ...
        
    
    
    class PullSubscription:
        """
        PullSubscription is a subscription that can fetch messages.
        """
        def __init__(self, js: JetStreamContext, sub: Subscription, stream: str, consumer: str, deliver: bytes) -> None:
            ...
        
        @property
        def pending_msgs(self) -> int:
            """
            Number of delivered messages by the NATS Server that are being buffered
            in the pending queue.
            """
            ...
        
        @property
        def pending_bytes(self) -> int:
            """
            Size of data sent by the NATS Server that is being buffered
            in the pending queue.
            """
            ...
        
        @property
        def delivered(self) -> int:
            """
            Number of delivered messages to this subscription so far.
            """
            ...
        
        async def unsubscribe(self) -> None:
            """
            unsubscribe destroys the inboxes of the pull subscription making it
            unable to continue to receive messages.
            """
            ...
        
        async def consumer_info(self) -> api.ConsumerInfo:
            """
            consumer_info gets the current info of the consumer from this subscription.
            """
            ...
        
        async def fetch(self, batch: int = ..., timeout: Optional[float] = ..., heartbeat: Optional[float] = ...) -> List[Msg]:
            """
            fetch makes a request to JetStream to be delivered a set of messages.

            :param batch: Number of messages to fetch from server.
            :param timeout: Max duration of the fetch request before it expires.
            :param heartbeat: Idle Heartbeat interval in seconds for the fetch request.

            ::

                import asyncio
                import nats

                async def main():
                    nc = await nats.connect()
                    js = nc.jetstream()

                    await js.add_stream(name='mystream', subjects=['foo'])
                    await js.publish('foo', b'Hello World!')

                    msgs = await sub.fetch(5)
                    for msg in msgs:
                      await msg.ack()

                    await nc.close()

                if __name__ == '__main__':
                    asyncio.run(main())
            """
            ...
        
    
    
    async def key_value(self, bucket: str) -> KeyValue:
        ...
    
    async def create_key_value(self, config: Optional[api.KeyValueConfig] = ..., **params) -> KeyValue:
        """
        create_key_value takes an api.KeyValueConfig and creates a KV in JetStream.
        """
        ...
    
    async def delete_key_value(self, bucket: str) -> bool:
        """
        delete_key_value deletes a JetStream KeyValue store by destroying
        the associated stream.
        """
        ...
    
    async def object_store(self, bucket: str) -> ObjectStore:
        ...
    
    async def create_object_store(self, bucket: str = ..., config: Optional[api.ObjectStoreConfig] = ..., **params) -> ObjectStore:
        """
        create_object_store takes an api.ObjectStoreConfig and creates a OBJ in JetStream.
        """
        ...
    
    async def delete_object_store(self, bucket: str) -> bool:
        """
        delete_object_store will delete the underlying stream for the named object.
        """
        ...
    


