FM / auth

/opt/imh-python/lib/python3.9/site-packages/zmq/auth/

/opt/imh-python/lib/python3.9/site-packages/zmq/auth

Name	Size	Mode	Actions
asyncio/	-	0755	rm
__pycache__/	-	0755	rm
base.py	14847	0644	edit dl rm
certs.py	4179	0644	edit dl rm
ioloop.py	1106	0644	edit dl rm
thread.py	7602	0644	edit dl rm
__init__.py	328	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/zmq/auth/ioloop.py (1106B)

"""ZAP Authenticator integrated with the tornado IOLoop.

.. versionadded:: 14.1
"""

# Copyright (C) PyZMQ Developers
# Distributed under the terms of the Modified BSD License.

from tornado import ioloop
from zmq.eventloop import zmqstream
from .base import Authenticator

class IOLoopAuthenticator(Authenticator):
    """ZAP authentication for use in the tornado IOLoop"""

def __init__(self, context=None, encoding='utf-8', log=None, io_loop=None):
        super(IOLoopAuthenticator, self).__init__(context, encoding, log)
        self.zap_stream = None
        self.io_loop = io_loop or ioloop.IOLoop.current()

def start(self):
        """Start ZAP authentication"""
        super(IOLoopAuthenticator, self).start()
        self.zap_stream = zmqstream.ZMQStream(self.zap_socket, self.io_loop)
        self.zap_stream.on_recv(self.handle_zap_message)

def stop(self):
        """Stop ZAP authentication"""
        if self.zap_stream:
            self.zap_stream.close()
            self.zap_stream = None
        super(IOLoopAuthenticator, self).stop()

__all__ = ['IOLoopAuthenticator']

"""ZAP Authenticator integrated with the tornado IOLoop. .. versionadded:: 14.1 """ # Copyright (C) PyZMQ Developers # Distributed under the terms of the Modified BSD License. from tornado import ioloop from zmq.eventloop import zmqstream from .base import Authenticator class IOLoopAuthenticator(Authenticator): """ZAP authentication for use in the tornado IOLoop""" def __init__(self, context=None, encoding='utf-8', log=None, io_loop=None): super(IOLoopAuthenticator, self).__init__(context, encoding, log) self.zap_stream = None self.io_loop = io_loop or ioloop.IOLoop.current() def start(self): """Start ZAP authentication""" super(IOLoopAuthenticator, self).start() self.zap_stream = zmqstream.ZMQStream(self.zap_socket, self.io_loop) self.zap_stream.on_recv(self.handle_zap_message) def stop(self): """Stop ZAP authentication""" if self.zap_stream: self.zap_stream.close() self.zap_stream = None super(IOLoopAuthenticator, self).stop() __all__ = ['IOLoopAuthenticator']