actually create/return the pipe
430913
function piper(to, cbPiper)
  {
    if(pipes[to]) return cbPiper(pipes[to]);

    // make a new pipe for this peering
    var pipe = pipes[to] = new Pipe('peer');
    pipe.to = to;
    pipe.path = {type:'peer',hn:to};

    // handle any peer delivery through the router
    function peer_send(packet, link, cbSend)
    {
      var router = mesh.index[to];
      if(!router) return cbSend('cannot peer to an unknown router: '+pipe.to);
      if(!router.x) return cbSend('cannot peer yet via this router: '+pipe.to);
      if(!link) return cbSend('requires link');

      // no packet means try to send our keys
      if(!packet)
      {
        Object.keys(mesh.keys).forEach(function(csid){
          if(link.csid && link.csid != csid) return; // if we know the csid, only send that key
          var json = {type:'peer',peer:link.hashname,c:router.x.cid()};
          var body = lob.encode(hashname.intermediates(mesh.keys), hashname.key(csid, mesh.keys));
          var attach = lob.encode({type:'link', csid:csid}, body);
          log.debug('sending peer key to',router.hashname,json,csid);
          router.x.send({json:json,body:attach});
        });
        return;
      }

      // if it's an encrypted channel packet, pass through direct to router
      if(packet.head.length == 0) return router.x.sending(packet);

      // otherwise we're always creating a new peer channel to carry the request
      var json = {type:'peer',peer:link.hashname,c:router.x.cid()};
      var body = lob.encode(packet);
      log.debug('sending peer handshake to',router.hashname,json,body);
      router.x.send({json:json,body:body});
      cbSend();
    }


    pipe.on('send', function(context,a1, a2, a3, a4){
      peer_send.call(context,a1,a2,a3,a4)
    })

    cbPiper(pipe);
  }
313469
def _self_pipe(self):
        
        import fcntl

        self._poll_fd, self._poll_send = os.pipe()
        for fd in [self._poll_fd, self._poll_send]:
            fl = fcntl.fcntl(fd, fcntl.F_GETFL)
            fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
455566
def pipe(self, cmd1, cmd2, quiet=False, estimated_size=None, **kwa):
        """"""
        if self.has_pv and not quiet:
            pv = "pv" if estimated_size is None else "pv --size {}".format(estimated_size)
            return self.shell("{} | {}| {}".format(cmd1, pv, cmd2), **kwa)
        else:
            return self.shell("{} | {}".format(cmd1, cmd2), **kwa)
1230648
function pipeFile (file, endCallback) {
    testPipeFile();
    testFunction(endCallback, 'endCallback');

    let readable = fs.createReadStream(file);

    readableEndCallback = endCallback;

    readable.once('error', throwAsyncError);
    readable.once('close', onReadableClose);

    readable.pipe(writable, { end: false });
}
1301557
function createStream (options) {
        options = options || {}

        var stream = through2.obj(function transform (chunk, encoding, cb) {
            stream.push(options.objectMode ? chunk : JSON.stringify(chunk)+'\n')
            cb()
        })
        stream.close = function close () {
            setImmediate(function () {
                stream.push(null)
                changes.unpipe(stream)
            })
            numStreams--
        }
        changes.pipe(stream)
        numStreams++
        return stream
    }
1039547
public static Pipe newPipe(MessageBufferInput in, boolean numeric) throws IOException
    {
        MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(in);
        return newPipe(unpacker, numeric);
    }
271016
function pipe(opts, cb) {
  if (typeof opts === 'function') {
    cb = opts;
    opts = {};
  }

  if (!cb) {
    cb = function () { }
  }

  opts = opts || {};

  var results = validate(opts);

  if (!hasErrors(results)) {
    return cb();
  }

  var err = new Error('Unable to validate dependencies');
  err.meta = results;
  return cb(err);
}
623998
def get_return_code(self, stderr=STDOUT):
        
        args = shlex.split(self.cmd)
        return call(args, stdout=PIPE, stderr=stderr)
800542
function (pipe) {
        return glob('*/', {cwd: pipe.contentDir})
            .then(function (categories) {
                categories.forEach(function (category) {
                    // Don't override already defined structures
                    if (pipe.content[category] !== undefined && pipe.content[category].type !== undefined) {
                        return;
                    }

                    pipe.content[category] = {
                        type: 'posts'
                    };
                });

                return Bluebird.resolve(pipe);
            });
    }
342720
function pipe(funcs) {
  return function pipeRewire(config) {
    const args = Array.prototype.slice.call(arguments, 1);
    (funcs || []).forEach(func => {
      if (func && typeof func === 'function') {
        // slice the first argument "config"
        config = func.apply(this, [config].concat(args));
      }
    });
    return config;
  };
}
1022053
def create (self, name, *args):
        """"""
        tab  = None
        defn = self.get(name, None)
        if defn:
            tab = FSWTab(defn, *args)
        return tab
647756
function recreate(makeStream, options) {
  if (options === undefined && isObject(makeStream)) {
    options = makeStream;
    makeStream = options.makeStream;
  }

  options = options || {};

  if (options.connect === undefined) {
    options.connect = true;
  }

  var connectedEvent = options.connectedEvent || 'open';

  var stream = through();
  stream.inverse = through();

  stream.inverse.pause();

  stream.write = function(chunk) {
    return stream.inverse.write(chunk);
  };

  stream.connect = function() {
    stream.underlying = makeStream();

    stream.inverse.pipe(stream.underlying, {end: false});

    stream.underlying.on('data', function(chunk) {
      return stream.emit('data', chunk);
    });

    stream.underlying.on(connectedEvent, function() {
      stream.backoff.reset();
      stream.inverse.resume();
      stream.emit('open');
      stream.emit('underlying-open');
    });

    stream.underlying.on('end', function() {
      stream.inverse.pause();
      if (!stream.preventBackoff) {
        stream.backoff.backoff();
      }
      stream.emit('underlying-end');
    });

    stream.underlying.on('error', function(e) {
      stream.inverse.pause();
      if (!stream.preventBackoff) {
        stream.backoff.backoff();
      }
      stream.emit('underlying-error');
    });
  };

  var origEnd = stream.end.bind(stream);

  stream.end = function() {
    stream.preventBackoff = true;
    stream.underlying.end();
    return origEnd();
  };

  stream.backoff = options.backoff || backoff.exponential(options);

  stream.backoff.on('backoff', function(num, delay) {
    stream.emit('backoff-schedule', num, delay);
  });

  stream.backoff.on('ready', function() {
    stream.connect();
    stream.emit('backoff-ready');
  });

  stream.backoff.on('fail', function() {
    stream.emit('backoff-fail');
  });


  if (options.connect) {
    stream.connect();
  }

  return stream;
}
312745
def source_pipe(self, source, ps=None):
        """"""

        if isinstance(source, string_types):
            source = self.source(source)

        source.dataset = self.dataset
        source._bundle = self

        iter_source, source_pipe = self._iterable_source(source, ps)

        if self.limited_run:
            source_pipe.limit = 500

        return iter_source, source_pipe
1126110
public static Pipe newPipe(byte[] data, int offset, int length, boolean numeric) throws IOException
    {
        ArrayBufferInput in = new ArrayBufferInput(data, offset, length);
        return newPipe(in, numeric);
    }
731870
def _pipe_stdio(cls, sock, stdin_isatty, stdout_isatty, stderr_isatty, handle_stdin):
    """"""
    stdio_writers = (
      (ChunkType.STDOUT, stdout_isatty),
      (ChunkType.STDERR, stderr_isatty)
    )
    types, ttys = zip(*(stdio_writers))

    @contextmanager
    def maybe_handle_stdin(want):
      if want:
        # TODO: Launching this thread pre-fork to handle @rule input currently results
        # in an unhandled SIGILL in `src/python/pants/engine/scheduler.py, line 313 in pre_fork`.
        # More work to be done here in https://github.com/pantsbuild/pants/issues/6005
        with NailgunStreamStdinReader.open(sock, stdin_isatty) as fd:
          yield fd
      else:
        with open('/dev/null', 'rb') as fh:
          yield fh.fileno()

    with maybe_handle_stdin(handle_stdin) as stdin_fd,\
         NailgunStreamWriter.open_multi(sock, types, ttys) as ((stdout_fd, stderr_fd), writer),\
         stdio_as(stdout_fd=stdout_fd, stderr_fd=stderr_fd, stdin_fd=stdin_fd):
      # N.B. This will be passed to and called by the `DaemonExiter` prior to sending an
      # exit chunk, to avoid any socket shutdown vs write races.
      stdout, stderr = sys.stdout, sys.stderr
      def finalizer():
        try:
          stdout.flush()
          stderr.flush()
        finally:
          time.sleep(.001)  # HACK: Sleep 1ms in the main thread to free the GIL.
          writer.stop()
          writer.join()
          stdout.close()
          stderr.close()
      yield finalizer
396162
def make_or_pipe(pipe):
    
    p1 = OrPipe(pipe)
    p2 = OrPipe(pipe)
    p1._partner = p2
    p2._partner = p1
    return p1, p2
301089
def create_pipeline(self, onetime=None):
        """"""
        utils.banner("Creating Pipeline")

        kwargs = {
            'app': self.app,
            'trigger_job': self.trigger_job,
            'prop_path': self.json_path,
            'base': None,
            'runway_dir': self.runway_dir,
        }

        pipeline_type = self.configs['pipeline']['type']

        if pipeline_type not in consts.ALLOWED_TYPES:
            raise NotImplementedError('Pipeline type "{0}" not permitted.'.format(pipeline_type))

        if not onetime:
            if pipeline_type == 'lambda':
                spinnakerpipeline = pipeline.SpinnakerPipelineLambda(**kwargs)
            elif pipeline_type == 's3':
                spinnakerpipeline = pipeline.SpinnakerPipelineS3(**kwargs)
            elif pipeline_type == 'datapipeline':
                spinnakerpipeline = pipeline.SpinnakerPipelineDataPipeline(**kwargs)
            elif pipeline_type == 'manual':
                spinnakerpipeline = pipeline.SpinnakerPipelineManual(**kwargs)
            else:
                # Handles all other pipelines
                spinnakerpipeline = pipeline.SpinnakerPipeline(**kwargs)
        else:
            spinnakerpipeline = pipeline.SpinnakerPipelineOnetime(onetime=onetime, **kwargs)

        spinnakerpipeline.create_pipeline()