<commit_before>#!/usr/bin/env python
# -- Content-Encoding: UTF-8 --
"""
Pelix remote services: ZooKeeper-based discovery and event notification

*Note:* This discovery package requires the ``kazoo`` package (available on
PyPI), and a ZooKeeper server.

:author: <NAME>
:copyright: Copyright 2020, <NAME>
:license: Apache License 2.0
:version: 1.0.0

..

    Copyright 2020 <NAME>

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
"""

# Standard library
import logging
import posixpath
import socket
import threading
import uuid

# ZooKeeper library
from kazoo.client import KazooClient, KazooState, EventType
from kazoo.exceptions import KazooException, NodeExistsError

# iPOPO decorators
from pelix.ipopo.decorators import (
    ComponentFactory,
    Requires,
    Provides,
    Invalidate,
    Validate,
    Property,
)

# Pelix utilities
import pelix.constants
from pelix.utilities import to_bytes, to_str
from pelix.threadpool import ThreadPool

# Remote services
from pelix.remote.edef_io import EDEFWriter, EDEFReader
import pelix.remote
import pelix.remote.beans as beans

# ------------------------------------------------------------------------------

# Module version
__version_info__ = (1, 0, 0)
__version__ = ".".join(str(x) for x in __version_info__)

# Documentation strings format
__docformat__ = "restructuredtext en"

_logger = logging.getLogger(__name__)
""" Module's logger """

FRAMEWORKS_ROOT = "/frameworks"
""" Path to the Z-node parent of frameworks definition Z-node """

ENDPOINTS_ROOT = "/endpoints"
""" Path to the Z-node containing the endpoints of each framework """

# ------------------------------------------------------------------------------


class ZooKeeperClient:
    """
    Handles basic ZooKeeper events
    """

    def __init__(self, zk_hosts, log_name="ZooKeeperClient", prefix=""):
        """
        :param zk_hosts: List of ZooKeepers hosts
        :param log_name: Name of the logger to use
        :param prefix: Prefix to all paths
        """
        self._zk = KazooClient(zk_hosts)
        self._zk.add_listener(self.__conn_listener)
        self.__prefix = prefix

        # Session state
        self.__connected = False
        self.__online = False
        self.__stop = False

        # Notification queue
        self._logger = logging.getLogger(log_name)
        self._queue = ThreadPool(1, 1, logname=log_name)

    @property
    def prefix(self):
        """
        Prefix to all ZooKeeper nodes
        """
        return self.__prefix

    @property
    def connected(self):
        """
        ZooKeeper client state: connected to the quorum
        """
        return self.__connected

    @property
    def online(self):
        """
        ZooKeeper client state: connected & online (session active)
        """
        return self.__online

    @property
    def stopped(self):
        """
        ZooKeeper client status (stop requested)
        """
        return self.__stop

    def __conn_listener(self, state):
        """
        Connection event listener

        :param state: The new connection state
        """
        if state == KazooState.CONNECTED:
            self.__online = True
            if not self.__connected:
                self.__connected = True
                self._logger.info("Connected to ZooKeeper")
                self._queue.enqueue(self.on_first_connection)
            else:
                self._logger.warning("Re-connected to ZooKeeper")
                self._queue.enqueue(self.on_client_reconnection)
        elif state == KazooState.SUSPENDED:
            self._logger.warning("Connection suspended")
            self.__online = False
        elif state == KazooState.LOST:
            self.__online = False
            self.__connected = False

            if self.__stop:
                self._logger.info("Disconnected from ZooKeeper (requested)")
            else:
                self._logger.warning("Connection lost")

    def start(self):
        """
        Starts the connection
        """
        self.__stop = False
        self._queue.start()
        self._zk.start()

    def stop(self):
        """
        Stops the connection
        """
        self.__stop = True
        self._queue.stop()
        self._zk.stop()

    @staticmethod
    def on_first_connection():
        """
        Called when the client is connected for the first time
        """
        pass

    @staticmethod
    def on_client_reconnection():
        """
        Called when the client is reconnected to the server
        """
        pass

    def __path(self, path):
        """
        Adds the prefix to the given path

        :param path: Z-Path
        :return: Prefixed Z-Path
        """
        if path.startswith(self.__prefix):
            return path

        return "{}{}".format(self.__prefix, path)

    def create(self, path, data, ephemeral=False, sequence=False):
        """
        Creates a ZooKeeper node

        :param path: Z-Path
        :param data: Node Content
        :param ephemeral: Ephemeral flag
        :param sequence: Sequential flag
        """
        return self._zk.create(
            self.__path(path), data, ephemeral=ephemeral, sequence=sequence
        )
<commit_msg>ZK: Added missing type hint import<commit_after>#!/usr/bin/env python
# -- Content-Encoding: UTF-8 --
"""
Pelix remote services: ZooKeeper-based discovery and event notification

*Note:* This discovery package requires the ``kazoo`` package (available on
PyPI), and a ZooKeeper server.

:author: <NAME>
:copyright: Copyright 2020, <NAME>
:license: Apache License 2.0
:version: 1.0.0

..

    Copyright 2020 <NAME>

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
"""

# Standard library
import logging
import posixpath
import socket
import threading
import uuid

# ZooKeeper library
from kazoo.client import KazooClient, KazooState, EventType
from kazoo.client import WatchedEvent  # pylint:disable=unused-import
from kazoo.exceptions import KazooException, NodeExistsError

# iPOPO decorators
from pelix.ipopo.decorators import (
    ComponentFactory,
    Requires,
    Provides,
    Invalidate,
    Validate,
    Property,
)

# Pelix utilities
import pelix.constants
from pelix.utilities import to_bytes, to_str
from pelix.threadpool import ThreadPool

# Remote services
from pelix.remote.edef_io import EDEFWriter, EDEFReader
import pelix.remote
import pelix.remote.beans as beans

# ------------------------------------------------------------------------------

# Module version
__version_info__ = (1, 0, 0)
__version__ = ".".join(str(x) for x in __version_info__)

# Documentation strings format
__docformat__ = "restructuredtext en"

_logger = logging.getLogger(__name__)
""" Module's logger """

FRAMEWORKS_ROOT = "/frameworks"
""" Path to the Z-node parent of frameworks definition Z-node """

ENDPOINTS_ROOT = "/endpoints"
""" Path to the Z-node containing the endpoints of each framework """

# ------------------------------------------------------------------------------


class ZooKeeperClient:
    """
    Handles basic ZooKeeper events
    """

    def __init__(self, zk_hosts, log_name="ZooKeeperClient", prefix=""):
        """
        :param zk_hosts: List of ZooKeepers hosts
        :param log_name: Name of the logger to use
        :param prefix: Prefix to all paths
        """
        self._zk = KazooClient(zk_hosts)
        self._zk.add_listener(self.__conn_listener)
        self.__prefix = prefix

        # Session state
        self.__connected = False
        self.__online = False
        self.__stop = False

        # Notification queue
        self._logger = logging.getLogger(log_name)
        self._queue = ThreadPool(1, 1, logname=log_name)

    @property
    def prefix(self):
        """
        Prefix to all ZooKeeper nodes
        """
        return self.__prefix

    @property
    def connected(self):
        """
        ZooKeeper client state: connected to the quorum
        """
        return self.__connected

    @property
    def online(self):
        """
        ZooKeeper client state: connected & online (session active)
        """
        return self.__online

    @property
    def stopped(self):
        """
        ZooKeeper client status (stop requested)
        """
        return self.__stop

    def __conn_listener(self, state):
        """
        Connection event listener

        :param state: The new connection state
        """
        if state == KazooState.CONNECTED:
            self.__online = True
            if not self.__connected:
                self.__connected = True
                self._logger.info("Connected to ZooKeeper")
                self._queue.enqueue(self.on_first_connection)
            else:
                self._logger.warning("Re-connected to ZooKeeper")
                self._queue.enqueue(self.on_client_reconnection)
        elif state == KazooState.SUSPENDED:
            self._logger.warning("Connection suspended")
            self.__online = False
        elif state == KazooState.LOST:
            self.__online = False
            self.__connected = False

            if self.__stop:
                self._logger.info("Disconnected from ZooKeeper (requested)")
            else:
                self._logger.warning("Connection lost")

    def start(self):
        """
        Starts the connection
        """
        self.__stop = False
        self._queue.start()
        self._zk.start()

    def stop(self):
        """
        Stops the connection
        """
        self.__stop = True
        self._queue.stop()
        self._zk.stop()

    @staticmethod
    def on_first_connection():
        """
        Called when the client is connected for the first time
        """

    @staticmethod
    def on_client_reconnection():
        """
        Called when the client is reconnected to the server
        """

    def __path(self, path):
        """
        Adds the prefix to the given path

        :param path: Z-Path
        :return: Prefixed Z-Path
        """
        if path.startswith(self.__prefix):
            return path

        return "{}{}".format(self.__prefix, path)

    def create(self, path, data, ephemeral=False, sequence=False):
        """
        Creates a ZooKeeper node

        :param path: Z-Path
        :param data: Node Content
        :param ephemeral: Ephemeral flag
        :param sequence: Sequential flag
        """
        return self._zk.create(
            self.__path(path), data, ephemeral=ephemeral, sequence=sequence
        )

{'fraction_non_alphanumeric': 0.11167864254177128, 'fraction_numerical': 0.003982339191411999, 'mean_word_length': 2.1280801516382346, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5882087', 'n_tokens_mistral': 3201, 'n_tokens_neox': 3035, 'n_words': 960}