Gecko [ goons4good.org ]

Linux server.edchosting.com 4.18.0-553.79.1.lve.el7h.x86_64 #1 SMP Wed Oct 15 16:34:46 UTC 2025 x86_64
LiteSpeed
Server IP : 75.98.162.185 & Your IP : 216.73.216.163
Domains : Cant Read [ /etc/named.conf ]
User : goons4good

Name	Size	Permission	Date	Action
__pycache__	[ DIR ]	drwxr-xr-x	2022-03-29 05:07
__init__.py	1.35 KB	-rw-r--r--	2018-02-10 08:02

Rename

"""ZAP Authenticator integrated with the asyncio IO loop.

.. versionadded:: 15.2
"""

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

import asyncio

import zmq
from zmq.asyncio import Poller
from ..base import Authenticator

class AsyncioAuthenticator(Authenticator):
    """ZAP authentication for use in the asyncio IO loop"""

def __init__(self, context=None, loop=None):
        super().__init__(context)
        self.loop = loop or asyncio.get_event_loop()
        self.__poller = None
        self.__task = None

@asyncio.coroutine
    def __handle_zap(self):
        while True:
            events = yield from self.__poller.poll()
            if self.zap_socket in dict(events):
                msg = yield from self.zap_socket.recv_multipart()
                self.handle_zap_message(msg)

def start(self):
        """Start ZAP authentication"""
        super().start()
        self.__poller = Poller()
        self.__poller.register(self.zap_socket, zmq.POLLIN)
        self.__task = asyncio.ensure_future(self.__handle_zap())

def stop(self):
        """Stop ZAP authentication"""
        if self.__task:
            self.__task.cancel()
        if self.__poller:
            self.__poller.unregister(self.zap_socket)
            self.__poller = None
        super().stop()

__all__ = ['AsyncioAuthenticator']