1  # Copyright (c) Microsoft Corporation. All rights reserved.
   2  # Licensed under the MIT License. See LICENSE in the project root
   3  # for license information.
   4  
   5  import codecs
   6  import os
   7  import threading
   8  
   9  from debugpy import launcher
  10  from debugpy.common import log
  11  
  12  
  13  class CaptureOutput(object):
  14      """Captures output from the specified file descriptor, and tees it into another
  15      file descriptor while generating DAP "output" events for it.
  16      """
  17  
  18      instances = {}
  19      """Keys are output categories, values are CaptureOutput instances."""
  20  
  21      def __init__(self, whose, category, fd, stream):
  22          assert category not in self.instances
  23          self.instances[category] = self
  24          log.info("Capturing {0} of {1}.", category, whose)
  25  
  26          self.category = category
  27          self._whose = whose
  28          self._fd = fd
  29          self._decoder = codecs.getincrementaldecoder("utf-8")(errors="surrogateescape")
  30  
  31          if stream is None:
  32              # Can happen if running under pythonw.exe.
  33              self._stream = None
  34          else:
  35              self._stream = stream.buffer
  36              encoding = stream.encoding
  37              if encoding is None or encoding == "cp65001":
  38                  encoding = "utf-8"
  39              try:
  40                  self._encode = codecs.getencoder(encoding)
  41              except Exception:
  42                  log.swallow_exception(
  43                      "Unsupported {0} encoding {1!r}; falling back to UTF-8.",
  44                      category,
  45                      encoding,
  46                      level="warning",
  47                  )
  48                  self._encode = codecs.getencoder("utf-8")
  49              else:
  50                  log.info("Using encoding {0!r} for {1}", encoding, category)
  51  
  52          self._worker_thread = threading.Thread(target=self._worker, name=category)
  53          self._worker_thread.start()
  54  
  55      def __del__(self):
  56          fd = self._fd
  57          if fd is not None:
  58              try:
  59                  os.close(fd)
  60              except Exception:
  61                  pass
  62  
  63      def _worker(self):
  64          while self._fd is not None:
  65              try:
  66                  s = os.read(self._fd, 0x1000)
  67              except Exception:
  68                  break
  69              if not len(s):
  70                  break
  71              self._process_chunk(s)
  72  
  73          # Flush any remaining data in the incremental decoder.
  74          self._process_chunk(b"", final=True)
  75  
  76      def _process_chunk(self, s, final=False):
  77          s = self._decoder.decode(s, final=final)
  78          if len(s) == 0:
  79              return
  80  
  81          try:
  82              launcher.channel.send_event(
  83                  "output", {"category": self.category, "output": s.replace("\r\n", "\n")}
  84              )
  85          except Exception:
  86              pass  # channel to adapter is already closed
  87  
  88          if self._stream is None:
  89              return
  90  
  91          try:
  92              s, _ = self._encode(s, "surrogateescape")
  93              size = len(s)
  94              i = 0
  95              while i < size:
  96                  written = self._stream.write(s[i:])
  97                  self._stream.flush()
  98                  if written == 0:
  99                      # This means that the output stream was closed from the other end.
 100                      # Do the same to the debuggee, so that it knows as well.
 101                      os.close(self._fd)
 102                      self._fd = None
 103                      break
 104                  i += written
 105          except Exception:
 106              log.swallow_exception("Error printing {0!r} to {1}", s, self.category)
 107  
 108  
 109  def wait_for_remaining_output():
 110      """Waits for all remaining output to be captured and propagated."""
 111      for category, instance in CaptureOutput.instances.items():
 112          log.info("Waiting for remaining {0} of {1}.", category, instance._whose)
 113          instance._worker_thread.join()
 114