How to use the imjoy.imjoySocketIO_client.transports.XHR_PollingTransport function in imjoy

To help you get started, we’ve selected a few imjoy examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github oeway / ImJoy-Engine / imjoy / imjoySocketIO_client / transports.py View on Github external
def __init__(self, http_session, is_secure, url, engineIO_session=None):
        super(XHR_PollingTransport, self).__init__(
            http_session, is_secure, url, engineIO_session
        )
        self._params = {"EIO": ENGINEIO_PROTOCOL, "transport": "polling"}
        if engineIO_session:
            self._request_index = 1
            self._kw_get = dict(timeout=engineIO_session.ping_timeout)
            self._kw_post = dict(
                timeout=engineIO_session.ping_timeout,
                headers={"content-type": "application/octet-stream"},
            )
            self._params["sid"] = engineIO_session.id
        else:
            self._request_index = 0
            self._kw_get = {}
            self._kw_post = {}
        http_scheme = "https" if is_secure else "http"
github oeway / ImJoy-Engine / imjoy / imjoySocketIO_client / __init__.py View on Github external
def _get_transport(self, transport_name):
        SelectedTransport = {
            "xhr-polling": XHR_PollingTransport,
            "websocket": WebsocketTransport,
        }[transport_name]
        return SelectedTransport(
            self._http_session, self._is_secure, self._url, self._engineIO_session
        )
github oeway / ImJoy-Engine / imjoy / imjoySocketIO_client / __init__.py View on Github external
def _get_engineIO_session(self):
        warning_screen = self._yield_warning_screen()
        for elapsed_time in warning_screen:
            transport = XHR_PollingTransport(
                self._http_session, self._is_secure, self._url
            )
            try:
                engineIO_packet_type, engineIO_packet_data = next(
                    transport.recv_packet()
                )
                break
            except (TimeoutError, ConnectionError) as e:
                if not self._wait_for_connection:
                    raise
                warning = Exception("[waiting for connection] %s" % e)
                warning_screen.throw(warning)
        assert engineIO_packet_type == 0  # engineIO_packet_type == open
        return parse_engineIO_session(engineIO_packet_data)