Append a new processor to the pipe
321105
def append_processor(self, proc, source_proc=None):
        ""
        if source_proc is None and len(self.processors):
            source_proc = self.processors[0]

        if source_proc and not isinstance(source_proc, Processor):
            raise TypeError('source_proc must be a Processor or None')

        if not isinstance(proc, Processor):
            raise TypeError('proc must be a Processor or None')

        if proc.type == 'decoder' and len(self.processors):
            raise ValueError('Only the first processor in a pipe could be a Decoder')

        # TODO : check if the processor is already in the pipe
        if source_proc:
            for child in self._graph.neighbors_iter(source_proc.uuid()):
                child_proc = self._graph.node[child]['processor']
                if proc == child_proc:
                    proc._uuid = child_proc.uuid()
                    proc.process_pipe = self
                    break
        if not self._graph.has_node(proc.uuid()):
            self.processors.append(proc)  # Add processor to the pipe
            self._graph.add_node(proc.uuid(), processor=proc, id=proc.id())
            if source_proc:
                self._graph.add_edge(self.processors[0].uuid(), proc.uuid(),
                                     type='audio_source')
            proc.process_pipe = self
            # Add an edge between each parent and proc
            for parent in proc.parents.values():
                self._graph.add_edge(parent.uuid(), proc.uuid(),
                                     type='data_source')
1193463
public static Pipe newPipe(byte[] data, int offset, int length) throws IOException
    {
        return newPipe(new ByteArrayInputStream(data, offset, length));
    }
1138306
def add_process(self, command=None, vsplit=False, start_directory=None):
        
        assert command is None or isinstance(command, six.text_type)
        assert start_directory is None or isinstance(start_directory, six.text_type)

        window = self.arrangement.get_active_window()

        pane = self._create_pane(window, command, start_directory=start_directory)
        window.add_pane(pane, vsplit=vsplit)
        pane.focus()
        self.invalidate()
128769
function Processor(options) {
    options = options || {};
    this.files = options.files || FILES;
    this.from  = options.from || DIR_FROM;
    this.to = options.to || DIR_TO;
    this.clear = options.clear || false;
}
178615
def append(self, cb):
        """"""
        self.callbacks.append(cb.callb)
        self.monitor = self.monitor | cb.callb.bit

        yield from self.pi._pigpio_aio_command(_PI_CMD_NB, self.handle,
                                               self.monitor)
1219790
@Override
	public Processor newProcessor(Processor p) {
		AMRulesAggregatorProcessor oldProcessor = (AMRulesAggregatorProcessor) p;
		Builder builder = new Builder(oldProcessor);
		AMRulesAggregatorProcessor newProcessor = builder.build();
		newProcessor.resultStream = oldProcessor.resultStream;
		newProcessor.statisticsStream = oldProcessor.statisticsStream;
		return newProcessor;
	}
47731
private static void addProcessors(final Processor processor, final String... keys) throws IllegalStateException {
        for (final String key : keys) {
            addProcessor(key, processor);
        }
    }
910149
private boolean setupPipelineForAppend(LocatedBlock lastBlock) throws IOException {
    if (nodes == null || nodes.length == 0) {
      String msg = "Could not get block locations. " +
          "Source file \"" + src
          + "\" - Aborting...";
      DFSClient.LOG.warn(msg);
      setLastException(new IOException(msg));
      closed = true;
      if (streamer != null) streamer.close();
      return false;
    }
    
    boolean success = createBlockOutputStream(nodes, dfsClient.clientName, false, true);
    long oldGenerationStamp = 
        ((LocatedBlockWithOldGS)lastBlock).getOldGenerationStamp();
    
    if (success) {
      // bump up the generation stamp in NN.
      Block newBlock = lastBlock.getBlock();
      Block oldBlock = new Block(newBlock.getBlockId(), newBlock.getNumBytes(), 
          oldGenerationStamp);
      dfsClient.namenode.updatePipeline(dfsClient.clientName, 
          oldBlock, newBlock, nodes);
    } else {
      DFSClient.LOG.warn("Fall back to block recovery process when trying" +
          " to setup the append pipeline for file " + src);
      // set the old generation stamp 
      block.setGenerationStamp(oldGenerationStamp);
      // fall back the block recovery
      while(processDatanodeError(true, true)) {
        try {
          Thread.sleep(1000);
        } catch (InterruptedException  e) {
          lastException = new IOException(e);
          break;
        }
      }
    }
    return success;
  }
1374021
def add_on_cpu(self, transforms):
        """"""
        if not isinstance(transforms, list):
            transforms = [transforms]
        self.cpu_transforms.extend(transforms or [])
        return self
868777
def add_process(self, name, cmd, quiet=False, env=None, cwd=None):
        
        assert name not in self._processes, "process names must be unique"
        proc = self._process_ctor(cmd,
                                  name=name,
                                  quiet=quiet,
                                  colour=next(self._colours),
                                  env=env,
                                  cwd=cwd)
        self._processes[name] = {}
        self._processes[name]['obj'] = proc

        # Update printer width to accommodate this process name
        self._printer.width = max(self._printer.width, len(name))

        return proc
239985
def wrap_existing_process(self, pid, stdout_read_fd, stderr_read_fd, port=None):
        
        stdout_read_file = os.fdopen(stdout_read_fd, 'rb')
        stderr_read_file = os.fdopen(stderr_read_fd, 'rb')
        stdout_streams, stderr_streams = self._get_stdout_stderr_streams()
        self._stdout_tee = io_wrap.Tee(stdout_read_file, *stdout_streams)
        self._stderr_tee = io_wrap.Tee(stderr_read_file, *stderr_streams)

        self.proc = Process(pid)
        self._run.pid = pid
        logger.info("wrapping existing process %i" % pid)

        try:
            self.init_run()
        except LaunchError as e:
            logger.exception("catostrophic launch error")
            wandb.termerror(str(e))
            util.sentry_exc(e)
            self._socket.launch_error()
            return

        if io_wrap.SIGWINCH_HANDLER is not None:
            # SIGWINCH_HANDLER (maybe) gets set in self.init_run()
            io_wrap.SIGWINCH_HANDLER.add_fd(stdout_read_fd)
            io_wrap.SIGWINCH_HANDLER.add_fd(stderr_read_fd)

        # Signal the main process that we're all hooked up
        logger.info("informing user process we are ready to proceed")
        self._socket.ready()

        self._sync_etc(headless=True)
1247634
def cli(ctx, amount, index, stage):
    """"""
    ctx.obj.say_green('Starting Streaming Pipe')
    res_pull = ctx.invoke(pull, amount=amount, index=index, stage=stage)
    res_tra = False
    if res_pull:
        # amount to transform can be less (or more)
        res_tra = ctx.invoke(
            transform, amount=amount, index=index, stage=stage)
    if res_tra:
        # amount to push can be less (or more)
        res_push = ctx.invoke(push, amount=amount, index=index, stage=stage)
    if res_pull and res_tra and res_push:
        ctx.obj.say_green('Streaming Pipe finsished')
        return True
    return False
1261197
def main():
    
    parser = argparse.ArgumentParser(
        description='A pipeline that generates analysis pipelines.')
    parser.add_argument('input', nargs='?',
                   help='A valid metapipe configuration file.')
    parser.add_argument('-o', '--output',
                   help='An output destination. If none is provided, the '
                   'results will be printed to stdout.',
                   default=sys.stdout)
    parser.add_argument('-t', '--temp',
                   help='A desired metapipe binary file. This is used to store '
                   'temp data between generation and execution. '
                   '(Default: "%(default)s")', default='.metapipe')
    parser.add_argument('-s', '--shell',
                   help='The path to the shell to be used when executing the '
                   'pipeline. (Default: "%(default)s)"',
                   default='/bin/bash')
    parser.add_argument('-r', '--run',
                   help='Run the pipeline as soon as it\'s ready.',
                   action='store_true')
    parser.add_argument('-n', '--name',
                   help='A name for the pipeline.',
                   default='')
    parser.add_argument('-j', '--job-type',
                   help='The destination for calculations (i.e. local, a PBS '
                   'queue on a cluster, etc).\nOptions: {}. '
                   '(Default: "%(default)s)"'.format(JOB_TYPES.keys()),
                   default='local')
    parser.add_argument('-p', '--max-jobs',
                   help='The maximum number of concurrent jobs allowed. '
                   'Defaults to maximum available cores.',
                   default=None)
    parser.add_argument('--report-type',
                   help='The output report type. By default metapipe will '
                   'print updates to the console. \nOptions: {}. '
                   '(Default: "%(default)s)"'.format(QUEUE_TYPES.keys()),
                   default='text')
    parser.add_argument('-v','--version',
                    help='Displays the current version of the application.',
                    action='store_true')
    args = parser.parse_args()

    if args.version:
        print('Version: {}'.format(__version__))
        sys.exit(0)

    try:
        with open(args.input) as f:
            config = f.read()
    except IOError:
        print('No valid config file found.')
        return -1

    run(config, args.max_jobs, args.output, args.job_type, args.report_type,
        args.shell, args.temp, args.run)
27471
def pipe_to_process(self, payload):
        """"""
        message = payload['input']
        key = payload['key']
        if not self.process_handler.is_running(key):
            return {'message': 'No running process for this key',
                    'status': 'error'}
        self.process_handler.send_to_process(message, key)
        return {'message': 'Message sent',
                'status': 'success'}
451319
def create_process(cmd, root_helper=None, addl_env=None, log_output=True):
    
    if root_helper:
        cmd = shlex.split(root_helper) + cmd
    cmd = map(str, cmd)

    log_output and LOG.info("Running command: %s", cmd)
    env = os.environ.copy()
    if addl_env:
        env.update(addl_env)

    obj = subprocess_popen(cmd, shell=False, stdin=subprocess.PIPE,
                           stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                           env=env)
    return obj, cmd
867609
def merge_results(self, other_processor):
        """"""
        if not isinstance(other_processor, self.__class__):
            raise ValueError("Can only extend with another %s instance."
                             % self.__class__.__name__)
        self.statements.extend(other_processor.statements)
        if other_processor.statements_sample is not None:
            if self.statements_sample is None:
                self.statements_sample = other_processor.statements_sample
            else:
                self.statements_sample.extend(other_processor.statements_sample)

        self._merge_json(other_processor.__statement_jsons,
                         other_processor.__evidence_counts)
        return
986515
def append(self, *other):
    
    self._data = it.chain(self._data, Stream(*other)._data)
    return self