# Path: catkin_tools/execution/io.py
# class IOBufferLogger(IOBufferContainer):
# 
#     """This is a logging class to be used instead of sys.stdout and sys.stderr
#     in FunctionStage operations.
# 
#     This class also generates `stdout` and `stderr` events.
#     """
# 
#     def __init__(self, label, job_id, stage_label, event_queue, log_path, *args, **kwargs):
#         IOBufferContainer.__init__(self, label, job_id, stage_label, event_queue, log_path)
# 
#     def out(self, data, end='\n'):
#         """
#         :type data: str
#         :type end: str
#         """
#         # Buffer the encoded data
#         data += end
#         encoded_data = self._encode(data)
#         self.stdout_buffer += encoded_data
#         self.interleaved_buffer += encoded_data
# 
#         # Save the encoded data
#         self.log_file.write(encoded_data)
# 
#         # Emit event with decoded Python str
#         self.event_queue.put(ExecutionEvent(
#             'STDOUT',
#             job_id=self.job_id,
#             stage_label=self.stage_label,
#             data=data))
# 
#     def err(self, data, end='\n'):
#         """
#         :type data: str
#         :type end: str
#         """
#         # Buffer the encoded data
#         data += end
#         encoded_data = self._encode(data)
#         self.stderr_buffer += encoded_data
#         self.interleaved_buffer += encoded_data
# 
#         # Save the encoded data
#         self.log_file.write(encoded_data)
# 
#         # Emit event with decoded Python str
#         self.event_queue.put(ExecutionEvent(
#             'STDERR',
#             job_id=self.job_id,
#             stage_label=self.stage_label,
#             data=data))
# 
# Path: catkin_tools/execution/io.py
# class IOBufferProtocol(IOBufferContainer, AsyncSubprocessProtocol):
# 
#     """An asyncio protocol that collects stdout and stderr.
# 
#     This class also generates `stdout` and `stderr` events.
# 
#     Since the underlying asyncio API constructs the actual protocols, this
#     class provides a factory method to inject the job and stage information
#     into the created protocol.
#     """
# 
#     def __init__(self, label, job_id, stage_label, event_queue, log_path, *args, **kwargs):
#         IOBufferContainer.__init__(self, label, job_id, stage_label, event_queue, log_path)
#         AsyncSubprocessProtocol.__init__(self, *args, **kwargs)
# 
#         self.intermediate_stdout_buffer = b''
#         self.intermediate_stderr_buffer = b''
# 
#     @staticmethod
#     def _split(data):
#         try:
#             last_break = data.rindex(b'\n') + 1
#             return data[0:last_break], data[last_break:]
#         except ValueError:
#             return b'', data
# 
#     def on_stdout_received(self, data):
#         """
#         :type data: encoded bytes
#         """
# 
#         data, self.intermediate_stdout_buffer = self._split(self.intermediate_stdout_buffer + data)
# 
#         self.stdout_buffer += data
#         self.interleaved_buffer += data
#         self.log_file.write(data)
# 
#         # Get the decoded Python str
#         decoded_data = self._decode(data)
# 
#         # Emit event with decoded Python str
#         self.event_queue.put(ExecutionEvent(
#             'STDOUT',
#             job_id=self.job_id,
#             stage_label=self.stage_label,
#             data=decoded_data))
# 
#     def on_stderr_received(self, data):
#         """
#         :type data: encoded bytes
#         """
# 
#         data, self.intermediate_stderr_buffer = self._split(self.intermediate_stderr_buffer + data)
# 
#         self.stderr_buffer += data
#         self.interleaved_buffer += data
#         self.log_file.write(data)
# 
#         # Get the decoded Python str
#         decoded_data = self._decode(data)
# 
#         # Emit event with decoded Python str
#         self.event_queue.put(ExecutionEvent(
#             'STDERR',
#             job_id=self.job_id,
#             stage_label=self.stage_label,
#             data=decoded_data))
# 
#     def on_process_exited2(self, returncode):
#         """
#         Dump anything remaining in the intermediate buffers.
#         """
# 
#         if len(self.intermediate_stdout_buffer) > 0:
#             self.on_stdout_received(self.intermediate_stdout_buffer + b'\n')
#         if len(self.intermediate_stderr_buffer) > 0:
#             self.on_stderr_received(self.intermediate_stderr_buffer + b'\n')


import os
import traceback
from shlex import quote as cmd_quote
from .io import IOBufferLogger
from .io import IOBufferProtocol
        self.label = str(label)
        self.logger_factory = logger_factory
        self.occupy_job = occupy_job
        self.repro = repro
        self.locked_resource = locked_resource
        self.early_termination_retcode = early_termination_retcode
        self.success_retcodes = success_retcodes

    def get_reproduction_cmd(self, verb, jid):
        """Get a command line to reproduce this stage with the proper environment."""
        return self.repro


class CommandStage(Stage):

    """Job stage that describes a system command.
    """

    def __init__(
            self,
            label,
            cmd,
            cwd=None,
            env=None,
            env_overrides=None,
            shell=False,
            emulate_tty=True,
            stderr_to_stdout=False,
            occupy_job=True,
            locked_resource=None,

# Path: catkin_tools/execution/io.py
# class IOBufferLogger(IOBufferContainer):
# 
#     """This is a logging class to be used instead of sys.stdout and sys.stderr
#     in FunctionStage operations.
# 
#     This class also generates `stdout` and `stderr` events.
#     """
# 
#     def __init__(self, label, job_id, stage_label, event_queue, log_path, *args, **kwargs):
#         IOBufferContainer.__init__(self, label, job_id, stage_label, event_queue, log_path)
# 
#     def out(self, data, end='\n'):
#         """
#         :type data: str
#         :type end: str
#         """
#         # Buffer the encoded data
#         data += end
#         encoded_data = self._encode(data)
#         self.stdout_buffer += encoded_data
#         self.interleaved_buffer += encoded_data
# 
#         # Save the encoded data
#         self.log_file.write(encoded_data)
# 
#         # Emit event with decoded Python str
#         self.event_queue.put(ExecutionEvent(
#             'STDOUT',
#             job_id=self.job_id,
#             stage_label=self.stage_label,
#             data=data))
# 
#     def err(self, data, end='\n'):
#         """
#         :type data: str
#         :type end: str
#         """
#         # Buffer the encoded data
#         data += end
#         encoded_data = self._encode(data)
#         self.stderr_buffer += encoded_data
#         self.interleaved_buffer += encoded_data
# 
#         # Save the encoded data
#         self.log_file.write(encoded_data)
# 
#         # Emit event with decoded Python str
#         self.event_queue.put(ExecutionEvent(
#             'STDERR',
#             job_id=self.job_id,
#             stage_label=self.stage_label,
#             data=data))
# 
# Path: catkin_tools/execution/io.py
# class IOBufferProtocol(IOBufferContainer, AsyncSubprocessProtocol):
# 
#     """An asyncio protocol that collects stdout and stderr.
# 
#     This class also generates `stdout` and `stderr` events.
# 
#     Since the underlying asyncio API constructs the actual protocols, this
#     class provides a factory method to inject the job and stage information
#     into the created protocol.
#     """
# 
#     def __init__(self, label, job_id, stage_label, event_queue, log_path, *args, **kwargs):
#         IOBufferContainer.__init__(self, label, job_id, stage_label, event_queue, log_path)
#         AsyncSubprocessProtocol.__init__(self, *args, **kwargs)
# 
#         self.intermediate_stdout_buffer = b''
#         self.intermediate_stderr_buffer = b''
# 
#     @staticmethod
#     def _split(data):
#         try:
#             last_break = data.rindex(b'\n') + 1
#             return data[0:last_break], data[last_break:]
#         except ValueError:
#             return b'', data
# 
#     def on_stdout_received(self, data):
#         """
#         :type data: encoded bytes
#         """
# 
#         data, self.intermediate_stdout_buffer = self._split(self.intermediate_stdout_buffer + data)
# 
#         self.stdout_buffer += data
#         self.interleaved_buffer += data
#         self.log_file.write(data)
# 
#         # Get the decoded Python str
#         decoded_data = self._decode(data)
# 
#         # Emit event with decoded Python str
#         self.event_queue.put(ExecutionEvent(
#             'STDOUT',
#             job_id=self.job_id,
#             stage_label=self.stage_label,
#             data=decoded_data))
# 
#     def on_stderr_received(self, data):
#         """
#         :type data: encoded bytes
#         """
# 
#         data, self.intermediate_stderr_buffer = self._split(self.intermediate_stderr_buffer + data)
# 
#         self.stderr_buffer += data
#         self.interleaved_buffer += data
#         self.log_file.write(data)
# 
#         # Get the decoded Python str
#         decoded_data = self._decode(data)
# 
#         # Emit event with decoded Python str
#         self.event_queue.put(ExecutionEvent(
#             'STDERR',
#             job_id=self.job_id,
#             stage_label=self.stage_label,
#             data=decoded_data))
# 
#     def on_process_exited2(self, returncode):
#         """
#         Dump anything remaining in the intermediate buffers.
#         """
# 
#         if len(self.intermediate_stdout_buffer) > 0:
#             self.on_stdout_received(self.intermediate_stdout_buffer + b'\n')
#         if len(self.intermediate_stderr_buffer) > 0:
#             self.on_stderr_received(self.intermediate_stderr_buffer + b'\n')


# Path: catkin_tools/execution/stages.py
import os
import traceback
from shlex import quote as cmd_quote
from .io import IOBufferLogger
from .io import IOBufferProtocol

        self.label = str(label)
        self.logger_factory = logger_factory
        self.occupy_job = occupy_job
        self.repro = repro
        self.locked_resource = locked_resource
        self.early_termination_retcode = early_termination_retcode
        self.success_retcodes = success_retcodes

    def get_reproduction_cmd(self, verb, jid):
        """Get a command line to reproduce this stage with the proper environment."""
        return self.repro


class CommandStage(Stage):

    """Job stage that describes a system command.
    """

    def __init__(
            self,
            label,
            cmd,
            cwd=None,
            env=None,
            env_overrides=None,
            shell=False,
            emulate_tty=True,
            stderr_to_stdout=False,
            occupy_job=True,
            locked_resource=None,

            logger_factory=IOBufferProtocol.factory,