# Path: src/TubeP.py
# class TubeP:
#     """A unidirectional communication channel 
#     using :class:`multiprocessing.Connection` for underlying implementation."""
# 
#     def __init__(self):
#         (self._conn1, 
#          self._conn2) = multiprocessing.Pipe(duplex=False)
# 
#     def put(self, data):
#         """Put an item on the tube."""
#         self._conn2.send(data)
# 
#     def get(self, timeout=None):
#         """Return the next available item from the tube.
# 
#         Blocks if tube is empty, until a producer for the tube puts an item on it."""
#         if timeout:
#             # Todo: Consider locking the poll/recv block.
#             # Otherwise, this method is not thread safe.
#             if self._conn1.poll(timeout):
#                 return (True, self._conn1.recv())
#             else:
#                 return (False, None)
#         return self._conn1.recv()


import multiprocessing
from .TubeP import TubeP
        self, 
        input_tube,      # Read task from the input tube.
        output_tubes,    # Send result on all the output tubes.
        num_workers,     # Total number of workers in the stage.
        disable_result,  # Whether to override any result with None.
        do_stop_task,    # Whether to call doTask() on "stop" request.
        ):
        """Create *num_workers* worker objects with *input_tube* and 
        an iterable of *output_tubes*. The worker reads a task from *input_tube* 
        and writes the result to *output_tubes*."""

        super(OrderedWorker, self).__init__()
        self._tube_task_input = input_tube
        self._tubes_result_output = output_tubes
        self._num_workers = num_workers

        # Serializes reading from input tube.
        self._lock_prev_input = None
        self._lock_next_input = None

        # Serializes writing to output tube.
        self._lock_prev_output = None
        self._lock_next_output = None

        self._disable_result = disable_result
        self._do_stop_task = do_stop_task

    @staticmethod
    def getTubeClass():
        """Return the tube class implementation."""

# Path: src/TubeP.py
# class TubeP:
#     """A unidirectional communication channel 
#     using :class:`multiprocessing.Connection` for underlying implementation."""
# 
#     def __init__(self):
#         (self._conn1, 
#          self._conn2) = multiprocessing.Pipe(duplex=False)
# 
#     def put(self, data):
#         """Put an item on the tube."""
#         self._conn2.send(data)
# 
#     def get(self, timeout=None):
#         """Return the next available item from the tube.
# 
#         Blocks if tube is empty, until a producer for the tube puts an item on it."""
#         if timeout:
#             # Todo: Consider locking the poll/recv block.
#             # Otherwise, this method is not thread safe.
#             if self._conn1.poll(timeout):
#                 return (True, self._conn1.recv())
#             else:
#                 return (False, None)
#         return self._conn1.recv()


# Path: src/OrderedWorker.py
import multiprocessing
from .TubeP import TubeP

        self, 
        input_tube,      # Read task from the input tube.
        output_tubes,    # Send result on all the output tubes.
        num_workers,     # Total number of workers in the stage.
        disable_result,  # Whether to override any result with None.
        do_stop_task,    # Whether to call doTask() on "stop" request.
        ):
        """Create *num_workers* worker objects with *input_tube* and 
        an iterable of *output_tubes*. The worker reads a task from *input_tube* 
        and writes the result to *output_tubes*."""

        super(OrderedWorker, self).__init__()
        self._tube_task_input = input_tube
        self._tubes_result_output = output_tubes
        self._num_workers = num_workers

        # Serializes reading from input tube.
        self._lock_prev_input = None
        self._lock_next_input = None

        # Serializes writing to output tube.
        self._lock_prev_output = None
        self._lock_next_output = None

        self._disable_result = disable_result
        self._do_stop_task = do_stop_task

    @staticmethod
    def getTubeClass():
        """Return the tube class implementation."""

        return TubeP