docs: def xenon_interactive_worker(
          machine, worker_config, input_queue=None, stderr_sink=None):
      """Uses Xenon to run a single remote interactive worker.
  
      Jobs are read from stdin, and results written to stdout.
  
      :param machine:
          Specification of the machine on which to run.
      :type machine: noodles.run.xenon.Machine
  
      :param worker_config:
          Job configuration. Specifies the command to be run remotely.
      :type worker_config: noodles.run.xenon.XenonJobConfig
      """
      if input_queue is None:
          input_queue = Queue()
  
      registry = worker_config.registry()
  
      @pull_map
      def serialise(obj):
          """Serialise incoming objects, yielding strings."""
          if isinstance(obj, JobMessage):
              print('serializing:', str(obj.node), file=sys.stderr)
          return (registry.to_json(obj, host='scheduler') + '\n').encode()
  
      @pull_map
      def echo(line):
          print('{} input: {}'.format(worker_config.name, line), file=sys.stderr)
          return line
  
      def do_iterate(source):
          for x in source():
              if x is EndOfQueue:
                  yield EndOfWork
                  return
              yield x
  
      job, output_stream = machine.scheduler.submit_interactive_job(
          worker_config.xenon_job_description,
          echo(lambda: serialise(lambda: do_iterate(input_queue.source))))
  
      @sink_map
      def echo_stderr(text):
          """Print lines."""
          for line in text.split('\n'):
              print("{}: {}".format(worker_config.name, line), file=sys.stderr)
  
      if stderr_sink is None:
          stderr_sink = echo_stderr()
  
      @pull
      def read_output(source):
          """Handle output from job, sending stderr data to given
          `stderr_sink`, passing on lines from stdout."""
          line_buffer = ""
          try:
              for chunk in source():
                  if chunk.stdout:
                      lines = chunk.stdout.decode().splitlines(keepends=True)
  
                      if not lines:
                          continue
  
                      if lines[0][-1] == '\n':
                          yield line_buffer + lines[0]
                          line_buffer = ""
                      else:
                          line_buffer += lines[0]
  
                      if len(lines) == 1:
                          continue
  
                      yield from lines[1:-1]
  
                      if lines[-1][-1] == '\n':
                          yield lines[-1]
                      else:
                          line_buffer = lines[-1]
  
                  if chunk.stderr:
                      for line in chunk.stderr.decode().split('\n'):
                          stripped_line = line.strip()
                          if stripped_line != '':
                              stderr_sink.send(stripped_line)
  
          except grpc.RpcError as e:
              return
  
      @pull_map
      def deserialise(line):
          result = registry.from_json(line, deref=False)
          return result
  
      return Connection(
          lambda: deserialise(lambda: read_output(lambda: output_stream)),
          input_queue.sink, aux=job)