TODO: Should this be included?  Do we need to convert it to a stream?
183870
function(func, opts) {
	opts = Object.assign({
		chunk_size: 25,
		retry: 2,
		retryDelay: 100,
		concurrency: 2,
		concurrency_delay: 100,
		combine: false,
		data_size: null
	}, opts || {});
	// logger.log("opts are ", opts);
	var records = [];
	var calls = 0;
	var completedCalls = 0;
	var requestEnd = false;

	var retries = 0;
	var errors = 0;
	var hadErrors = false;
	var batches = 0;
	var delaying = false;

	function sendAvailable() {
		var sendSize;
		if (records.length > 0 && retries <= opts.retry && completedCalls == calls && !delaying) {
			if (errors == 0) { //let's reset because last round completed successfully
				batches++;
				if (!hadErrors) {
					retries = 0;
				}
				hadErrors = false;
				logger.log(`-------------------New Batch #${batches}----------------`);
				if (opts.chunk_size < 10 || opts.concurrency > 25) {
					logger.log(`chunking ${opts.chunk_size} - ${opts.concurrency} times`);

				}
			} else {
				logger.log(`-------------------Retrying: ${errors} records failed, retrying in ${opts.retryDelay * retries}ms, retry #${opts.retry - (opts.retry - retries) + 1}----------------`);
				retries++;
				errors = 0;
				hadErrors = true;
				delaying = true;
				setTimeout(function() {
					delaying = false;
					sendAvailable();
				}, opts.retryDelay * retries);
				return;
			}
			if (retries > opts.retry) {
				checkDone();
				return;
			}
			while (records.length > 0 && completedCalls > calls - opts.concurrency) {
				calls++;
				var dataSizeBased = false;
				if (opts.data_size) {
					sendSize = 0;
					var runningSize = 0;
					for (var i = 0; i < opts.chunk_size && i < records.length; i++) {
						var r = records[i];
						runningSize += r.size;
						if (runningSize > opts.data_size) {
							dataSizeBased = true;
							break;
						}
						sendSize++;
					}
				} else {
					sendSize = opts.chunk_size;
				}
				if (opts.combine) {
					var items = records.splice(0, sendSize);
					var toProcess = [];
					var size = 0;
					var groupStart = 0;

					for (var i = 0; i < items.length; i++) {
						var item = items[i];
						if (item.size + size >= opts.record_size) {
							logger.log(`grouping items from ${groupStart+1} to ${i} of ${items.length} of size: ${size}`);
							toProcess.push(items.slice(groupStart, i).map((e) => {
								return e.record;
							}).join(''));
							groupStart = i;
							size = item.size;
						} else {
							size += item.size;
						}
					}
					if (groupStart != items.length) {
						logger.log(`grouping items from ${groupStart+1} to ${items.length} of ${items.length} of size: ${size}`);
						toProcess.push(items.slice(groupStart, items.length).map((e) => {
							return e.record;
						}).join(''));
					}

				} else {
					var toProcess = records.splice(0, sendSize).map(function(e) {
						return e.record;
					});
				}

				if (toProcess.length > 0) {
					if (opts.chunk_size >= 10 && opts.concurrency <= 25) {
						logger.log(`chunking ${toProcess.length} records (${dataSizeBased?'Data Size':'Count Size'})`);
					}
					func(toProcess, function(err, unprocessedItems) {
						if (err) {
							logger.log(`Records not processed, ${unprocessedItems.length}`);

							process.nextTick(function() {
								//Don't want to add the records or change completed calls until after the current While loop is done...otherwise a nasty infinite loop could happen
								completedCalls++;
								records = unprocessedItems.map(function(e) {
									var size;
									if (!size) {
										if (typeof e === "string") {
											size = Buffer.byteLength(e);
										} else {
											size = Buffer.byteLength(JSON.stringify(e));
										}
									}
									return {
										size: size,
										record: e
									};
								}).concat(records);
								errors += unprocessedItems.length;
								setTimeout(sendAvailable, opts.concurrency_delay);
							});
						} else if (records.length) {
							completedCalls++;
							setTimeout(sendAvailable, opts.concurrency_delay);
						} else {
							completedCalls++;
							sendAvailable();
						}
					});
				} else {
					completedCalls++;
				}
			}
		} else {
			checkDone();
		}
	}

	function checkDone() {
		if (requestEnd !== false && completedCalls == calls && (records.length == 0 || retries >= opts.retry)) {
			if (records.length > 0) {
				requestEnd("Cannot process all the entries", records.length);
				requestEnd = false;
			} else {
				requestEnd(null, []);
				requestEnd = false;
			}
		}
	}

	return {
		add: function(item) {
			requestEnd = false;
			var size;
			if (!size) {
				if (typeof item === "string") {
					size = Buffer.byteLength(item);
				} else {
					size = Buffer.byteLength(JSON.stringify(item));
				}
			}

			if (opts.record_size && size > opts.record_size) {
				logger.error("record size is too large", size, opts.record_size);
			} else if (opts.data_size && size > opts.data_size) {
				logger.error("data size is too large");
			} else {
				records.push({
					size: size,
					record: item
				});
				if (records.length >= opts.chunk_size * opts.concurrency) {
					sendAvailable();
				}
			}
		},
		end: function(callback) {
			requestEnd = callback;
			sendAvailable();
		}
	};
}
857230
function write(id, chunk) {
        // They want to write to our real stream
        var stream = streams[id];
        if (!stream) return;
        stream.write(chunk);
    }
759067
function addListenersToStream(listeners, stream) {
    var evt, callback;
    if (listeners) {
        for (evt in listeners) {
            if (listeners.hasOwnProperty(evt)) {
                callback = listeners[evt];
                stream.on(evt, callback);
            }
        }
    }
}
566861
private void dumpInputStream(){
        //dump request body
        exchange.startBlocking();
        InputStream inputStream = exchange.getInputStream();
        try {
            if(config.isMaskEnabled() && inputStream.available() != -1) {
                this.bodyContent = Mask.maskJson(inputStream, "requestBody");
            } else {
                try {
                    this.bodyContent = StringUtils.inputStreamToString(inputStream, UTF_8);
                } catch (IOException e) {
                    logger.error(e.toString());
                }
            }
        } catch (IOException e) {
            logger.error("undertow inputstream error:" + e.getMessage());
        }
    }
540942
function Stream( value, options ) {
		if ( !arguments.length ) {
			throw new Error( 'Stream()::insufficient input arguments. Must provide a value to append.' );
		}
		if ( arguments.length < 2 ) {
			options = {};
		}
		if ( !( this instanceof Stream ) ) {
			return new Stream( value, options );
		}
		var err = validate( options );
		if ( err ) {
			throw err;
		}

		Transform.call( this, options );
		this._val = value.toString();
		this._destroyed = false;

		return this;
	}
159247
function streamFile(stats) {
  if (!invokedHtaccess.call(
    this,
    200,
    "onstreamfile",
    stats
  )) {
    var
      stream = fs.createReadStream(this.path, emptyObjet),
      acceptEncoding = contentEncoding(
        this.request.headers['accept-encoding']
      )
    ;
    commonStream["Content-Type"] = polpetta_type(this.ext);
    commonStream["Content-Length"] = stats.size;
    // TODO: avoid if already binary
    if (acceptEncoding) {
      //http://nodejs.org/api/zlib.html
      commonStream["Content-Encoding"] = acceptEncoding;
      stream = stream.pipe(acceptEncoding === "deflate" ?
        zlib.createDeflate() : zlib.createGzip()
      );
    }
    this.response.writeHead(200, commonStream);
    stream.on("error", internalServerError.bind(this));
    stream.pipe(this.response);
  }
}
484300
function readableStream(stream) {
  return new Promise(function(resolve, reject) {
    var chunks = {
      data: []
    };

    function onData(chunk) {
      if (!chunks.type) {
        chunks.type = chunkTypes[typeof chunk];
      }

      chunks.data.push(chunk);
    }

    function onEnd() {
      if (chunks.type === chunkTypes.object) {
        resolve(Buffer.concat(chunks.data));
      }
      else {
        resolve(chunks.data.join(""));
      }
    }

    function onError(error) {
      reject(error);
    }

    stream
      .on("error", onError)
      .on("data", onData)
      .on("end", onEnd);
  });
}
417502
function OSMStream(options) {
  if (!(this instanceof OSMStream)) return new OSMStream(options);

  Transform.call(this, options);

  var primaryElements = ['node', 'way', 'relation']; // emit-ables
  var subElementName = ['tag', 'nd', 'member'];

  // size of bytes read from source so far
  this.bytesRead = 0;
  // size of bytes written to stream so far
  this.bytesWritten = 0;

  // initialize parser and its event handlers
  this._parser = new expat.Parser();

  this._parser.on('pipe', (function onParserPipe(src) {
    logger('_parser received pipe from ', src.path || '<unknown pipe>');
  }).bind(this));

  this._parser.on('startElement', (function onParserStartElement(name, attrs) {
    var funcname = ['__parse_', name].join('');
    if ((primaryElements.indexOf(name) > -1 && (OSMStream.prototype[funcname])) ||
      (subElementName.indexOf(name) > -1 && this._currentElement !== null && (OSMStream.prototype[funcname]))) {
      OSMStream.prototype[funcname].call(this, attrs);
    }
  }).bind(this));

  this._parser.on('endElement', (function onParserEndElement(name) {
    if ((primaryElements.indexOf(name) < 0)) return;

    var clone = extend({}, this._currentElement);
    this._currentElement = null;

    var pushData = function pushData(data) {
      if (!data) return;

      if (this.listeners('writeable').length > 0) {
        this.emit('writeable', data, (function onWriteable(data) {
          this.__push(data);
        }).bind(this));
      } else {
        this.__push(data);
      }

    }.bind(this);

    if (this.listeners(name).length > 0) {
      this.emit(name, clone, (function onElement(data) {
        pushData(data);
      }.bind(this)));
    } else {
      pushData(clone);
    }
  }).bind(this));

  this._parser.on('error', (function onParserError(err) {
    this.emit('error', err);
  }).bind(this));

  this._parser.on('end', (function onParserEnd() {
  }).bind(this));

  this.on('pipe', (function onPipe(src) {
    if (src === this._parser) return;
    // attaching stream pipe to parser
    logger('Received pipe from ', src.toString(), ', attaching pipe to _parser');
    src.pipe(this._parser);
    this._parser.pipe(this);
  }).bind(this));

}
631948
function appendStream(streams, s) {
    assert.arrayOfObject(streams, 'streams');
    assert.object(s, 'stream');

    if (s instanceof Stream) {
        streams.push({
            raw: false,
            stream: s
        });
    } else {
        assert.optionalBool(s.raw, 'stream.raw');
        assert.object(s.stream, 'stream.stream');
        streams.push(s);
    }
}
1324840
function streamToString(callback) {
  var contents = '';

  return through(
    function (chunk, enc, next) {
      contents += chunk.toString('utf8');
      next();
    },
    function (next) { // flush function
      callback.bind(this)(contents, next);
    }
  );
}
583055
function transformChunk(chunk, encoding, done) {
    if (chunk.isNull()) return done();
    if (chunk.isStream()) return this.emit('error', new PluginError(logFlag, 'Streaming not supported'));
    if (
      !(chunk.isMarkdown = options.regexpMarkdown.test(chunk.relative)) &&
      !(chunk.isTemplate = options.regexpTemplate.test(chunk.relative)) &&
      !(chunk.isHtml = options.regexpHtml.test(chunk.relative))
    ) {
      return options.passThrough ? done(null, chunk) : done();
    }
    transformChunkData(chunk);
    if (chunk.data.draft && process.env.NODE_ENV !== 'development') return done();
    if (buffer.hasOwnProperty(chunk.relative)) {
      logDuplicate(chunk);
      return done();
    }
    transformChunkContents.call(this, chunk);
    buffer[chunk.relative] = chunk;
    done(null, chunk);
  }
495243
function stream(nodeStream, linebreak) {
        linebreak = (typeof linebreak !== 'undefined' ? linebreak : '\n');
        
		var mw = core.base();
		mw.enabled = true;
		mw.stream = nodeStream;
		mw.linebreak = linebreak;
		mw.writeln = function (line) {
			if (mw.enabled) {
				mw.stream.write(line + linebreak);
			}
		};
		mw.toString = function () {
			return '<miniwrite-stream>';
		};
		return mw;
	}
1037000
function StreamReader(read) {

	var tail = '', messages = new Array();

	/**
	 * Parsa new data.
	 * @param {String} data New data from the stream
	 */
	function parse(data, initExpected) {
		var lines, i, last;
		data = tail + data;
		data = data.replace(/\r*\n+/g, "\n");
		lines = data.split("\n");
		last = lines.length - 1;
		for (i = 0; i < last; i++) {
			messages.push(read(lines[i], initExpected));
		}
		tail = lines[lines.length - 1];
	}

	/**
	 * Pop the next message
	 */
	function pop() {
		return messages.shift();
  	}

	/**
	 * Return if it has no more messages
	 */
	function isEmpty() {
		return messages.length == 0;
  	}

	// Public methods
	that = {
		parse: parse,
		pop: pop,
		isEmpty: isEmpty
	};


	return that;
}
1360723
@Override
  public StreamImpl stream()
    throws IOException
  {
    if (_streamImpl == null) {
      _streamImpl = new SocketStream();
    }

    _streamImpl.init(_s);

    return _streamImpl;
  }
553636
function () {
    var write = function (node, enc, cb) {
      node.value = encoder.encode(node.value, dag.valueEncoding)
      stream.node(node, cb)
    }

    stream.emit('live')
    pipe(dag.createReadStream({since: changes, live: true}), through.obj(write))
  }
316224
async function* streamToGenerator(stream) {
  let promise;

  stream.on('data', data => promise.resolve(data));
  stream.once('end', () => promise.resolve(null));

  stream.on('error', error => {
    stream.removeAllListeners();
    promise.reject(error);
  });

  while (true) {
    promise = defer();

    yield promise;
  }
}
148252
def add_input_stream(self, datastream, data_type, options, data):
        

        url = self.get_add_input_data_url(datastream, options)
        form_data = {
            'files': {
                'data': (
                    Utils.random_string(10)+('.json' if data_type == 'json' else '.csv'),
                    data,
                    'text/plain;charset=UTF-8',
                    {'Expires': '0'}
                )
            }
        }
        response = self.http.upstream(url, form_data)
        return response