STATE_NEW = 0 STATE_OPEN = 1 STATE_CLOSING = 2 STATE_CLOSED = 3 FRAME_OPEN = "o" FRAME_CLOSE = "c" FRAME_MESSAGE = "a" FRAME_MESSAGE_BLOB = "a1" FRAME_HEARTBEAT = "h" class SessionIsAcquired(SockjsException):
    """Session is acquired.""" class SessionIsClosed(SockjsException):
    """Session is closed.""" MSG_CLOSE = 3 MSG_MESSAGE = 2 def close_frame(code, reason):
    return FRAME_CLOSE + json.dumps([code, reason], **kwargs) def message_frame(message):
    return FRAME_MESSAGE + json.dumps([message], **kwargs) def messages_frame(messages):
    return FRAME_MESSAGE + json.dumps(messages, **kwargs) ENCODING = "utf-8"
STATE_NEW = 0
STATE_OPEN = 1
STATE_CLOSING = 2
STATE_CLOSED = 3
FRAME_OPEN = "o"
FRAME_CLOSE = "c"
FRAME_MESSAGE = "a"
FRAME_MESSAGE_BLOB = "a1"
FRAME_HEARTBEAT = "h"
IFRAME_HTML = """<!DOCTYPE html>
<html>
<head>
<meta http-equiv="X-UA-Compatible" content="IE=edge" />
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
  <script src="%s"></script>
  <script>
    document.domain = document.domain;
    SockJS.bootstrap_iframe();
  </script>
</head>
<body>
  <h2>Don't panic!</h2>
  <p>This is a SockJS hidden iframe. It's used for cross domain magic.</p>
</body>
</html>
""".strip()
IFRAME_MD5 = hashlib.md5(IFRAME_HTML.encode()).hexdigest()
ENCODING = "utf-8"
MSG_OPEN = 1
MSG_MESSAGE = 2
MSG_CLOSE = 3
MSG_CLOSED = 4
    def dthandler(obj):
def dumps(data):
def close_frame(code, reason):
def message_frame(message):
def messages_frame(messages):
    def tp(self):
class SockjsMessage(collections.namedtuple("SockjsMessage", ["type", "data"])):
import warnings
import asyncio
import collections
import logging
from datetime import datetime, timedelta
from asyncio import ensure_future
from .protocol import STATE_NEW, STATE_OPEN, STATE_CLOSING, STATE_CLOSED
from .protocol import FRAME_OPEN, FRAME_CLOSE
from .protocol import FRAME_MESSAGE, FRAME_MESSAGE_BLOB, FRAME_HEARTBEAT
from .exceptions import SessionIsAcquired, SessionIsClosed
from .protocol import MSG_CLOSE, MSG_MESSAGE
from .protocol import close_frame, message_frame, messages_frame
from .protocol import SockjsMessage, OpenMessage, ClosedMessage

    async def _remote_messages(self, messages):
        self._tick()

        for msg in messages:
            log.debug("incoming message: %s, %s", self.id, msg[:200])
            try:
                await self.handler(SockjsMessage(MSG_MESSAGE, msg), self)
            except Exception:
                log.exception("Exception in message handler.")

    def expire(self):
        """Manually expire a session."""
        self.expired = True

    def send(self, msg):
        """send message to client."""
        assert isinstance(msg, str), "String is required"

        if self._debug:
            log.info("outgoing message: %s, %s", self.id, str(msg)[:200])

        if self.state != STATE_OPEN:
            return

        self._feed(FRAME_MESSAGE, msg)

    def send_frame(self, frm):
        """send message frame to client."""
        if self._debug:
            log.info("outgoing message: %s, %s", self.id, frm[:200])

        if self.state != STATE_OPEN:
            return

        self._feed(FRAME_MESSAGE_BLOB, frm)

    def close(self, code=3000, reason="Go away!"):
        """close session"""
        if self.state in (STATE_CLOSING, STATE_CLOSED):
            return

        if self._debug:
            log.debug("close session: %s", self.id)

        self.state = STATE_CLOSING
        self._feed(FRAME_CLOSE, (code, reason))


_marker = object()


class SessionManager(dict):
    """A basic session manager."""

    _hb_handle = None  # heartbeat event loop timer
    _hb_task = None  # gc task

    def __init__(
        self,
        name,
        app,
        handler,
        heartbeat=25.0,
        timeout=timedelta(seconds=5),
        debug=False,
    ):
        self.name = name
        self.route_name = "sockjs-url-%s" % name
        self.app = app
        self.handler = handler
        self.factory = Session
        self.acquired = {}
        self.sessions = []
        self.heartbeat = heartbeat
        self.timeout = timeout
        self.debug = debug

    def route_url(self, request):
        return request.route_url(self.route_name)

    @property
    def started(self):
        return self._hb_handle is not None

    def start(self):
        if not self._hb_handle:
            loop = asyncio.get_event_loop()
            self._hb_handle = loop.call_later(self.heartbeat, self._heartbeat)

    def stop(self):
        if self._hb_handle is not None:
            self._hb_handle.cancel()
            self._hb_handle = None
        if self._hb_task is not None:
            self._hb_task.cancel()
            self._hb_task = None

    def _heartbeat(self):
        if self._hb_task is None:
            self._hb_task = ensure_future(self._heartbeat_task())

    async def _heartbeat_task(self):
        sessions = self.sessions

        if sessions:
            now = datetime.now()

            idx = 0
            while idx < len(sessions):
                session = sessions[idx]

                session._heartbeat()

                if session.expires < now:
                    # Session is to be GC'd immedietely
                    if session.id in self.acquired:
                        await self.release(session)
                    if session.state == STATE_OPEN:
                        await session._remote_close()

                    if session.state == STATE_CLOSING: