EventQueueEmptyEventHandler

Complexity

Total Complexity

15

Size/Duplication

Total Lines	87
Duplicated Lines	0 %

Importance

Changes	1
Bugs	0	Features	0

2 Methods

Rating	Name	Duplication	Size	Complexity
A	init()	0	6	1
F	handle()	0	67	14

"""
Enarksh

Copyright 2013-2016 Set Based IT Consultancy

Licence MIT
"""
import fcntl
import os
import select
import signal

import zmq

from enarksh.event.Event import Event
from enarksh.message.Message import Message

The name message does not seem to exist in module enarksh.

The import enarksh.message.Message could not be resolved.

class EventQueueEmptyEventHandler:
    """
    An event handler for an empty event queue.
    """
    _wake_up_pipe = None
    """
    The wakeup file descriptor when a signal is received.

    :type: None|(int,int)
    """

    # ------------------------------------------------------------------------------------------------------------------
    @classmethod
    def init(cls):

This method should have a docstring.

        cls._wake_up_pipe = os.pipe()
        fcntl.fcntl(cls._wake_up_pipe[0], fcntl.F_SETFL, os.O_NONBLOCK)

        signal.set_wakeup_fd(EventQueueEmptyEventHandler._wake_up_pipe[1])

    # ------------------------------------------------------------------------------------------------------------------
    @staticmethod
    def handle(_event, _event_data, spawner):
        """
        Handles an empty event queue event.
        

Trailing whitespace

        :param  _event: The event.
        :param * _event_data: Not used.
        :param enarksh.spawner.Spawner.Spawner spawner: The spawner.
        """
        del _event, _event_data

        # List with all file descriptors for reading.
        # Add the file descriptor for waking up select when a signal has been received.
        read = [EventQueueEmptyEventHandler._wake_up_pipe[0]]

        # Add the sockets for incoming messages to the list of read file descriptors.
        zmq_fds = set()
        for socket in Message.message_controller.end_points.values():
            if socket.type in [zmq.PULL, zmq.REP]:

The Module zmq does not seem to have a member named PULL.

The Module zmq does not seem to have a member named REP.

                zmq_fd = socket.get(zmq.FD)

The Module zmq does not seem to have a member named FD.

                read.append(zmq_fd)
                zmq_fds.add(zmq_fd)

        # Add the job handlers to the list of read file descriptors.
        for pid, job_handler in spawner.job_handlers.items():

The variable pid seems to be unused.

            fd_stdout = job_handler.stdout
            if fd_stdout >= 0:
                read.append(fd_stdout)
            fd_stderr = job_handler.stderr
            if fd_stderr >= 0:
                read.append(fd_stderr)

        # Unblock interrupts.
        signal.pthread_sigmask(signal.SIG_UNBLOCK, {signal.SIGCHLD, signal.SIGHUP})

        try:
            # Wait for a fd becomes available for read or wait for an interrupt.
            read, _, _ = select.select(read, [], [])

        except InterruptedError:
            # Ignore Interrupted system call errors (EINTR) in the select call.
            # Note: In Python 3.5 EINTR will not occur any more. Nevertheless, we must only allow interrupts here to
            # prevent unwanted side effect of the signal handlers.
            read = []

        # Block all interrupts to prevent interrupted system call errors (EINTR).
        signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGCHLD, signal.SIGHUP})

        zmq_event = False
        for fd in read:
            if fd in zmq_fds:
                # fd of the message queue is ready to receive data.
                zmq_event = True
            elif fd == EventQueueEmptyEventHandler._wake_up_pipe[0]:
                os.read(EventQueueEmptyEventHandler._wake_up_pipe[0], 512)
            else:
                # fd of one or more job handlers are ready to receive data.
                for job_handler in spawner.job_handlers.values():
                    if fd == job_handler.stdout:
                        job_handler.read(fd)
                    if fd == job_handler.stderr:
                        job_handler.read(fd)

        if zmq_event:
            spawner.zmq_incoming_message_event.fire()
        else:
            Event.event_controller.event_queue_empty.fire()  # XXX test queue is empty

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