# 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
        cmd_str = ' '.join([cmd_quote(t) for t in self.async_execute_process_kwargs['cmd']])

        # Define the command to run the subcommand
        env_cmd = 'catkin env -si {} {}'.format(
            env_overrides_formatted,
            cmd_str)

        # Return the full command
        return 'cd {}; {} | {}; cd -'.format(
            self.async_execute_process_kwargs['cwd'],
            get_env_cmd,
            env_cmd)


class FunctionStage(Stage):

    """Job stage that describes a python function.

    :param label: The label for the stage
    :param function: A python function which returns 0 on success

    Functions must take the arguments:
        - logger
        - event_queue
    """

    def __init__(
            self,
            label,
            function,

# 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

        cmd_str = ' '.join([cmd_quote(t) for t in self.async_execute_process_kwargs['cmd']])

        # Define the command to run the subcommand
        env_cmd = 'catkin env -si {} {}'.format(
            env_overrides_formatted,
            cmd_str)

        # Return the full command
        return 'cd {}; {} | {}; cd -'.format(
            self.async_execute_process_kwargs['cwd'],
            get_env_cmd,
            env_cmd)


class FunctionStage(Stage):

    """Job stage that describes a python function.

    :param label: The label for the stage
    :param function: A python function which returns 0 on success

    Functions must take the arguments:
        - logger
        - event_queue
    """

    def __init__(
            self,
            label,
            function,

            logger_factory=IOBufferLogger.factory,