Drain the stream into the destination stream
485555
private function getSaveToBody(
        RequestInterface $request,
        StreamInterface $stream
    ) {
        if ($saveTo = $request->getConfig()['save_to']) {
            // Stream the response into the destination stream
            $saveTo = is_string($saveTo)
                ? new Stream(Utils::open($saveTo, 'r+'))
                : Stream::factory($saveTo);
        } else {
            // Stream into the default temp stream
            $saveTo = Stream::factory();
        }

        Utils::copyToStream($stream, $saveTo);
        $saveTo->seek(0);
        $stream->close();

        return $saveTo;
    }
578499
protected void drainToLoad(long targetLoad, int time, boolean useHandoff)
    {
        final double startingLoad = myLoad();
        double currentLoad = startingLoad;
        List<String> drainList = new LinkedList<String>();
        Set<String> eligibleToDrop = new LinkedHashSet<String>(cluster.myWorkUnits);
        eligibleToDrop.removeAll(cluster.workUnitsPeggedToMe);

        for (String workUnit : eligibleToDrop) {
            if (currentLoad <= targetLoad) {
                break;
            }
            double workUnitLoad = cluster.getWorkUnitLoad(workUnit);

            if (workUnitLoad > 0 && (currentLoad - workUnitLoad) > targetLoad) {
                drainList.add(workUnit);
                currentLoad -= workUnitLoad;
            }
        }

        int drainInterval = (int) (((double) config.drainTime / drainList.size()) * 1000);
        TimerTask drainTask = buildDrainTask(drainList, drainInterval, useHandoff, currentLoad);

        if (!drainList.isEmpty()) {
            LOG.info("Releasing work units over {} seconds. Current load: {}. Target: {}. Releasing: {}",
                    time, startingLoad, targetLoad, Strings.mkstring(drainList, ", "));
            cluster.schedule(drainTask, 0, TimeUnit.SECONDS);
        }
    }
1250572
def FromStream(cls, stream, auto_transfer=True, total_size=None, **kwds):
        """"""
        return cls(stream, auto_transfer=auto_transfer, total_size=total_size,
                   **kwds)
1306282
protected void drain() {
        Message msg;
        while((msg=queue.poll()) != null)
            addAndSendIfSizeExceeded(msg);
        _sendBundledMessages();
    }
1360118
def drain(self):
        
        try:
            unlock = self.stick.acquire()
            return self.stick.drain()
        finally:
            unlock()
399339
def flush(self):
        
        self.acquire()
        try:
            if self.outStream and hasattr(self.outStream, "flush"):
                self.outStream.flush()
            if self.errStream and hasattr(self.errStream, "flush"):
                self.errStream.flush()
        finally:
            self.release()
34983
function sink( stream, clbk ) {
		var actual = [];
		stream.on( 'data', function onData( data ) {
			actual.push( data );
		});
		stream.once( 'end', function onEnd() {
			clbk( null, actual );
		});
		stream.once( 'error', function onError( error ) {
			clbk( error );
		});
	}
791469
private void fillOutputStream(OutputStream stream, int length)
			throws AsnException, IOException {
		byte[] dataBucket = new byte[DATA_BUCKET_SIZE];
		int readCount;

		while (length != 0) {
			int cnt = length < DATA_BUCKET_SIZE ? length : DATA_BUCKET_SIZE;
			readCount = read(dataBucket, 0, cnt);
			if (readCount < cnt)
				throw new AsnException("input stream has reached the end");
			stream.write(dataBucket, 0, readCount);
			length -= readCount;
		}
	}
1191496
function toStream(input) {
  let streamType = isStream(input);
  if (streamType === 'node') {
    return nodeToWeb(input);
  } else if (streamType) {
    return input;
  }
  return new ReadableStream({
    start(controller) {
      controller.enqueue(input);
      controller.close();
    }
  });
}
570175
function (inputStream, outputStream, tempPath, tempPath2, err, callback) {
      async.parallel({
        unlink: function(cb) {
          if (!tempPath) {
            return callback(null);
          }
          removeLocalBlob(tempPath, cb);
        },

        unlink2: function(cb) {
          if (!tempPath2) {
            return callback(null);
          }
          removeLocalBlob(tempPath2, cb);
        },

        closeReadStream: function(cb) {
          inputStream.destroy();
          cb();
        },

        closeWriteStream: function(cb) {
          outputStream.destroy();
          cb();
        }
      }, cleanupError => {
        if (err) {
          return callback(err);
        }
        return callback(cleanupError);
      });
    }
924287
public static void transferStream(Reader in, Writer out)
   {
       try {
           char[] cbuf = new char[1000];
           int iLen = 0;
           while ((iLen = in.read(cbuf, 0, cbuf.length)) > 0)
           {   // Write the entire file to the output buffer
               out.write(cbuf, 0, iLen);
           }
           in.close();
       } catch (MalformedURLException ex)  {
           ex.printStackTrace();
       } catch (IOException ex) {
           ex.printStackTrace();
       }
   }
1246745
function FileOutputStream(istream, destinationFilename, mode=false){
    // wrap into Promise
    return new Promise(function(resolve, reject){

        // file output stream
        let ostream = null;

        // on complete - chmod
        const onComplete = function(){
            // file mode defined ?
            if (mode){
                // set mode of destination file
                _fs.chmod(destinationFilename, mode, function(err){
                    if (err){
                        reject(err);
                    }else{
                        resolve();
                    }
                });
            }else{
                // finish
                resolve();
            }
        }

        // cleanup
        const onError = function(e){
            // detach finish listener!
            if (ostream){
                ostream.removeListener('finish', onComplete);
            }
            
            // close streams
            if (istream){
                istream.destroy();
            }
            if (ostream){
                ostream.end();
            }

            // throw error
            reject(e);
        }

        // add additional error listener to input stream
        istream.on('error', onError);

        // open write stream
        ostream = _fs.createWriteStream(destinationFilename);
        ostream.on('error', onError);
        ostream.on('finish', onComplete);

        // pipe in to out
        istream.pipe(ostream);
    });
}
962564
public int drainTo(@NotNull ByteBufQueue dest) {
		int size = 0;
		while (hasRemaining()) {
			ByteBuf buf = take();
			dest.add(buf);
			size += buf.readRemaining();
		}
		return size;
	}
1178869
void data(boolean outFinished, int streamId, Buffer source, boolean flush) {
    Preconditions.checkNotNull(source, "source");

    OkHttpClientStream stream = transport.getStream(streamId);
    if (stream == null) {
      // This is possible for a stream that has received end-of-stream from server (but hasn't sent
      // end-of-stream), and was removed from the transport stream map.
      // In such case, we just throw away the data.
      return;
    }

    OutboundFlowState state = state(stream);
    int window = state.writableWindow();
    boolean framesAlreadyQueued = state.hasPendingData();
    int size = (int) source.size();

    if (!framesAlreadyQueued && window >= size) {
      // Window size is large enough to send entire data frame
      state.write(source, size, outFinished);
    } else {
      // send partial data
      if (!framesAlreadyQueued && window > 0) {
        state.write(source, window, false);
      }
      // Queue remaining data in the buffer
      state.enqueue(source, (int) source.size(), outFinished);
    }

    if (flush) {
      flush();
    }
  }
1076289
public static void transfer(InputStream in, OutputStream out, TransferCallback cb) throws IOException
    {
        byte[] bytes = new byte[TRANSFER_BUFFER];
        int count;
        while ((count = in.read(bytes)) != -1)
        {
            out.write(bytes, 0, count);
            if (cb != null)
            {
                cb.bytesTransferred(bytes, count);
                if (cb.isCancelled())
                {
                    break;
                }
            }
        }
    }
7053
function through(transform, flush, objectMode) {
  const stream = new TransformStream({objectMode})
  stream._transform = transform || pass
  if (flush) stream._flush = flush
  return stream
}
644340
public static long transfer(final ReadableByteChannel source, final long count, final ByteBuffer throughBuffer, final WritableByteChannel sink) throws IOException {
        long total = 0L;
        while (total < count) {
            throughBuffer.clear();
            if (count - total < throughBuffer.remaining()) {
                throughBuffer.limit((int) (count - total));
            }

            try {
                long res = source.read(throughBuffer);
                if (res <= 0) {
                    return total == 0L ? res : total;
                }
            } finally {
                throughBuffer.flip();

            }
            while (throughBuffer.hasRemaining()) {
                long res = sink.write(throughBuffer);
                if (res <= 0) {
                    return total;
                }
                total += res;
            }
        }
        return total;
    }