<commit_before>"""Experimental FAUCET event notification."""

#### THIS API IS EXPERIMENTAL.
#### Discuss with faucet-dev list before relying on this API,
#### review http://www.hyrumslaw.com/.
#### It is subject to change without notice.

# TODO: events are currently schema-less. This is to facilitate rapid prototyping, and will change.
# TODO: not all cases where a notified client fails or could block, have been tested.

# Copyright (C) 2013 Nippon Telegraph and Telephone Corporation.
# Copyright (C) 2015 <NAME>, <NAME> and <NAME>.
# Copyright (C) 2015 Research and Education Advanced Network New Zealand Ltd.
# Copyright (C) 2015--2017 The Contributors
#
# 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.

import json
import os
import queue
import socket
import time

from ryu.lib import hub
from ryu.lib.hub import StreamServer


class FaucetExperimentalEventNotifier(object):
    """Event notification, via Unix domain socket."""

    def __init__(self, socket_path, metrics, logger):
        if socket_path:
            if os.path.exists(os.path.dirname(socket_path)):
                self.socket_path = socket_path
            else:
                self.socket_path = '/var/run/faucet/faucet.sock'
        else:
            self.socket_path = socket_path
        self.metrics = metrics
        self.logger = logger
        self.event_q = queue.Queue(120)
        self.event_id = 0

    def start(self):
        """Start socket server."""
        if self.socket_path:
            if os.path.exists(self.socket_path):
                os.remove(self.socket_path)
            return hub.spawn(
                StreamServer((self.socket_path, None), self._loop).serve_forever)
        return None

    def _loop(self, sock, _addr):
        """Serve events."""
        self.logger.info('event client connected')
        while True:
            while not self.event_q.empty():
                event = self.event_q.get()
                event_bytes = bytes('\n'.join((json.dumps(event), '')).encode('UTF-8'))
                try:
                    sock.sendall(event_bytes)
                except (socket.error, IOError) as err:
                    self.logger.info('event client disconnected' % err)
                    return
            hub.sleep(0.1)

    def notify(self, dp_id, dp_name, event_dict):
        """Notify of an event."""
        assert isinstance(event_dict, dict)
        self.event_id += 1
        event = {
            'version': 1,
            'time': time.time(),
            'dp_id': dp_id,
            'dp_name': dp_name,
            'event_id': self.event_id,
        }
        for header_key in list(event):
            assert header_key not in event_dict
        event.update(event_dict)
        if self.socket_path:
            self.metrics.faucet_event_id.set(event['event_id'])
            if self.event_q.full():
                self.logger.warning('event notify queue full')
            else:
                self.event_q.put(event)<commit_msg>Add additional sanity checking over state of event socket_path.<commit_after>"""Experimental FAUCET event notification."""

#### THIS API IS EXPERIMENTAL.
#### Discuss with faucet-dev list before relying on this API,
#### review http://www.hyrumslaw.com/.
#### It is subject to change without notice.

# TODO: events are currently schema-less. This is to facilitate rapid prototyping, and will change.
# TODO: not all cases where a notified client fails or could block, have been tested.

# Copyright (C) 2013 Nippon Telegraph and Telephone Corporation.
# Copyright (C) 2015 <NAME>, <NAME> and <NAME>.
# Copyright (C) 2015 Research and Education Advanced Network New Zealand Ltd.
# Copyright (C) 2015--2017 The Contributors
#
# 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.

import json
import os
import queue
import socket
import time

from ryu.lib import hub
from ryu.lib.hub import StreamServer


class FaucetExperimentalEventNotifier(object):
    """Event notification, via Unix domain socket."""

    def __init__(self, socket_path, metrics, logger):
        if socket_path:
            if os.path.exists(os.path.dirname(socket_path)):
                self.socket_path = socket_path
            else:
                self.socket_path = '/var/run/faucet/faucet.sock'
        else:
            self.socket_path = socket_path
        self.metrics = metrics
        self.logger = logger
        self.event_q = queue.Queue(120)
        self.event_id = 0
        self.thread = None

    def start(self):
        """Start socket server."""
        if self.socket_path:
            socket_dir = os.path.dirname(self.socket_path)
            if not os.path.exists(socket_dir):
                try:
                    os.makedirs(socket_dir)
                except (PermissionError) as err: # pytype: disable=name-error
                    self.logger.error('Unable to create event socket directory: %s', err)
                    return None
            if os.path.exists(self.socket_path):
                try:
                    os.remove(self.socket_path)
                except (PermissionError) as err: # pytype: disable=name-error
                    self.logger.error('Unable to remove old socket: %s', err)
                    return None
            if os.access(socket_dir, os.R_OK | os.W_OK | os.X_OK):
                self.thread = hub.spawn(
                    StreamServer((self.socket_path, None), self._loop).serve_forever)
                return self.thread
            else:
                self.logger.error('Incorrect permissions set on socket directory %s', socket_dir)
                return None

        return None

    def _loop(self, sock, _addr):
        """Serve events."""
        self.logger.info('event client connected')
        while True:
            while not self.event_q.empty():
                event = self.event_q.get()
                event_bytes = bytes('\n'.join((json.dumps(event), '')).encode('UTF-8'))
                try:
                    sock.sendall(event_bytes)
                except (socket.error, IOError) as err:
                    self.logger.info('event client disconnected' % err)
                    return
            hub.sleep(0.1)

    def notify(self, dp_id, dp_name, event_dict):
        """Notify of an event."""
        assert isinstance(event_dict, dict)
        self.event_id += 1
        event = {
            'version': 1,
            'time': time.time(),
            'dp_id': dp_id,
            'dp_name': dp_name,
            'event_id': self.event_id,
        }
        for header_key in list(event):
            assert header_key not in event_dict
        event.update(event_dict)
        if self.thread:
            self.metrics.faucet_event_id.set(event['event_id'])
            if self.event_q.full():
                self.logger.warning('event notify queue full')
            else:
                self.event_q.put(event)
{'fraction_non_alphanumeric': 0.09743394308943089, 'fraction_numerical': 0.008384146341463415, 'mean_word_length': 2.184870550161812, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, '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': '7040362', 'n_tokens_mistral': 2212, 'n_tokens_neox': 2093, 'n_words': 686}