///////////////////////////////////
// blockStream functions
////////////////////////////////////
550756
func newBlockStream(rootDir string, startFileNum int, startOffset int64, endFileNum int) (*blockStream, error) {
	startFileStream, err := newBlockfileStream(rootDir, startFileNum, startOffset)
	if err != nil {
		return nil, err
	}
	return &blockStream{rootDir, startFileNum, endFileNum, startFileStream}, nil
}
917126
function BlockEncoder(schema, opts) {
  opts = opts || {};

  stream.Duplex.call(this, {
    allowHalfOpen: true, // To support async compressors.
    writableObjectMode: true
  });

  var type;
  if (types.Type.isType(schema)) {
    type = schema;
    schema = undefined;
  } else {
    // Keep full schema to be able to write it to the header later.
    type = types.Type.forSchema(schema);
  }

  this._schema = schema;
  this._type = type;
  this._writeValue = function (tap, val) {
    try {
      this._type._write(tap, val);
    } catch (err) {
      this.emit('error', err);
    }
  };
  this._blockSize = opts.blockSize || 65536;
  this._tap = new Tap(utils.newBuffer(this._blockSize));
  this._codecs = opts.codecs;
  this._codec = opts.codec || 'null';
  this._blockCount = 0;
  this._syncMarker = opts.syncMarker || new utils.Lcg().nextBuffer(16);
  this._queue = new utils.OrderedQueue();
  this._pending = 0;
  this._finished = false;
  this._needHeader = false;
  this._needPush = false;

  this._metadata = opts.metadata || {};
  if (!MAP_BYTES_TYPE.isValid(this._metadata)) {
    throw new Error('invalid metadata');
  }

  var codec = this._codec;
  this._compress = (this._codecs || BlockEncoder.getDefaultCodecs())[codec];
  if (!this._compress) {
    throw new Error(f('unsupported codec: %s', codec));
  }

  if (opts.omitHeader !== undefined) { // Legacy option.
    opts.writeHeader = opts.omitHeader ? 'never' : 'auto';
  }
  switch (opts.writeHeader) {
    case false:
    case 'never':
      break;
    case undefined: // Backwards-compatibility (eager default would be better).
    case 'auto':
      this._needHeader = true;
      break;
    default:
      this._writeHeader();
  }

  this.on('finish', function () {
    this._finished = true;
    if (this._blockCount) {
      this._flushChunk();
    } else if (this._finished && this._needPush) {
      // We don't need to check `_isPending` since `_blockCount` is always
      // positive after the first flush.
      this.push(null);
    }
  });
}
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);
  });
}
497684
function singleStringCoding(stream, string, on_finish, on_progress) {
  on_progress = on_progress || function() {};
  on_finish = on_finish || function() {};

  // possibly our input is an array of byte integers
  // or a typed array
  if (!Buffer.isBuffer(string))
    string = Buffer.from(string);

  var deferred = {}, failed = false;

  stream.once('error', function(err) {
    failed = true;
    on_finish(null, err);
  });

  // emulate Promise.defer()
  deferred.promise = new Promise(function(resolve, reject) {
    deferred.resolve = resolve;
    deferred.reject = reject;
  });

  // Since using the Promise API is optional, generating unhandled
  // rejections is not okay.
  deferred.promise.catch(noop);

  stream.once('error', function(e) {
    deferred.reject(e);
  });

  var buffers = [];

  stream.on('data', function(b) {
    buffers.push(b);
  });

  stream.once('end', function() {
    var result = Buffer.concat(buffers);

    if (!failed) {
      on_progress(1.0);
      on_finish(result);
    }

    if (deferred)
      deferred.resolve(result);
  });

  on_progress(0.0);

  stream.end(string);

  if (deferred)
    return deferred.promise;
}
980115
function nodeStream(node) {
    var result = [];
    (function (node, offset) {
      for (var i = 0; i < node.childNodes.length; i++) {
        if (node.childNodes[i].nodeType == 3)
          offset += node.childNodes[i].nodeValue.length;
        else if (node.childNodes[i].nodeName == 'BR')
          offset += 1;
        else if (node.childNodes[i].nodeType == 1) {
          result.push({
            event: 'start',
            offset: offset,
            node: node.childNodes[i]
          });
          offset = arguments.callee(node.childNodes[i], offset);
          result.push({
            event: 'stop',
            offset: offset,
            node: node.childNodes[i]
          });
        }
      }
      return offset;
    })(node, 0);
    return result;
  }
415005
function Parser_findDefaultInlineStreamEnd(stream) {
      var E = 0x45, I = 0x49, SPACE = 0x20, LF = 0xA, CR = 0xD;
      var startPos = stream.pos, state = 0, ch, i, n, followingBytes;
      while ((ch = stream.getByte()) !== -1) {
        if (state === 0) {
          state = (ch === E) ? 1 : 0;
        } else if (state === 1) {
          state = (ch === I) ? 2 : 0;
        } else {
          assert(state === 2);
          if (ch === SPACE || ch === LF || ch === CR) {
            // Let's check the next five bytes are ASCII... just be sure.
            n = 5;
            followingBytes = stream.peekBytes(n);
            for (i = 0; i < n; i++) {
              ch = followingBytes[i];
              if (ch !== LF && ch !== CR && (ch < SPACE || ch > 0x7F)) {
                // Not a LF, CR, SPACE or any visible ASCII character, i.e.
                // it's binary stuff. Resetting the state.
                state = 0;
                break;
              }
            }
            if (state === 2) {
              break;  // Finished!
            }
          } else {
            state = 0;
          }
        }
      }
      return ((stream.pos - 4) - startPos);
    }
1058129
function streamOutput( callback ) {
	return function(err, filenames) {
		if (err) {
			callback(err, null);
		} else {
			// Create read streams from each input
			var streams = []
			for (var i=0; i<filenames.suffix.length; i++)
				streams.push([
					fs.createReadStream( filenames.basename + filenames.suffix[i]),
					filenames.suffix[i]
				]);
			// Callback with streams
			callback(err, streams);
		}
	}
}
458738
private void triggerAsyncCaching(BlockInStream stream) throws IOException {
    boolean cache = ReadType.fromProto(mOptions.getOptions().getReadType()).isCache();
    boolean overReplicated = mStatus.getReplicationMax() > 0
        && mStatus.getFileBlockInfos().get((int) (getPos() / mBlockSize))
        .getBlockInfo().getLocations().size() >= mStatus.getReplicationMax();
    cache = cache && !overReplicated;
    // Get relevant information from the stream.
    WorkerNetAddress dataSource = stream.getAddress();
    long blockId = stream.getId();
    if (cache && (mLastBlockIdCached != blockId)) {
      WorkerNetAddress worker;
      if (mPassiveCachingEnabled && mContext.hasLocalWorker()) { // send request to local worker
        worker = mContext.getLocalWorker();
      } else { // send request to data source
        worker = dataSource;
      }
      try {
        // Construct the async cache request
        long blockLength = mOptions.getBlockInfo(blockId).getLength();
        AsyncCacheRequest request =
            AsyncCacheRequest.newBuilder().setBlockId(blockId).setLength(blockLength)
                .setOpenUfsBlockOptions(mOptions.getOpenUfsBlockOptions(blockId))
                .setSourceHost(dataSource.getHost()).setSourcePort(dataSource.getDataPort())
                .build();
        BlockWorkerClient blockWorker = mContext.acquireBlockWorkerClient(worker);
        try {
          blockWorker.asyncCache(request);
          mLastBlockIdCached = blockId;
        } finally {
          mContext.releaseBlockWorkerClient(worker, blockWorker);
        }
      } catch (Exception e) {
        LOG.warn("Failed to complete async cache request for block {} at worker {}: {}", blockId,
            worker, e.getMessage());
      }
    }
  }
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;
	}
1001712
function IncomingStream(access_token, userid, groupids) {
    this.access_token = access_token;
    this.userid = userid;
    this.groupids = groupids;

    this.client = new WebSocketClient();
    this.connection = null;
    this.socketId = 1;

    this.state = STATE_DISCONNECTED;

    var self = this;

    var connectionInitHelper = function(connection) {

      connection.on('close', function(reasonCode, description) {
        self.emit('status', 'Websocket Disconnected.', description);
        this.connection = null;
        self.setState(STATE_DISCONNECTED);        
      });

      connection.on('error', function(error) {
        self.handleError("Websocket experienced error.", error);
        self.setState(STATE_DISCONNECTED);
      })
        
      connection.on('message', function(msg) {
        if (msg.type === 'utf8' && msg.utf8Data) {
          
          self.emit('status', 'Received:', msg);
          each(JSON.parse(msg.utf8Data), function(data) {

            if (data["channel"] == "/meta/handshake") {
              self.handshakeResponse(data);
            } else if (data["channel"] == "/meta/subscribe") {
              self.subscribeResponse(data);
            } else {
              self.emit('message', data);
            }

          });

        } else {
          self.emit('error','Received a non-utf8 message.', msg);
        }
      });

      return connection;

    }

    this.client.on('connect', function(connection) {
      self.setState(STATE_PENDING);  
      self.connection = connectionInitHelper(connection);
      self.emit('status', 'Websocket Connected');
      self.handshake();
    });

    this.client.on('connectFailed', function(err) {
      self.setState(STATE_DISCONNECTED);
      self.connection = null;
      self.emit('error', 'Connection Failed', err);
    });


}
1279406
function BlockMap() {

  if( !(this instanceof BlockMap) )
    return new BlockMap()

  this.signature = 0x00000000
  this.version = 0x00000000
  this.sectorNumber = 0x0000000000000000
  this.sectorCount = 0x0000000000000000
  this.dataOffset = 0x0000000000000000
  this.buffersNeeded = 0x00000000
  this.blockDescriptorCount = 0x00000000
  this.reserved1 = 0x00000000
  this.reserved2 = 0x00000000
  this.reserved3 = 0x00000000
  this.reserved4 = 0x00000000
  this.reserved5 = 0x00000000
  this.reserved6 = 0x00000000
  this.checksum = new UDIF.Checksum()
  this.blockCount = 0x00000000
  this.blocks = []

}
662142
function multiread(stream, get) {
    if (stream._multiread === undefined) {
        stream._multiread = stream._multiread || [];
        stream._multireading = stream._multireading || 0;
        stream._multireadmax = stream._multireadmax || concurrency;
    }
    if (stream._multiread.length) {
        push();
    } else {
        stream.once('_push', push);
    }
    while ((stream._multiread.length + stream._multireading) < stream._multireadmax) {
        stream._multireading++;
        if (!get(done)) break;
    }

    function done(item) {
        stream._multireading--;
        if (item === null && !stream._multireading) {
            stream.emit('_push');
        } else if (item) {
            stream._multiread.push(item);
            stream.emit('_push');
        }
    }

    function push() {
        if (stream._multiread.length) {
            stream.push(stream._multiread.shift());
        } else {
            stream.push(null);
        }
    }
}
97919
function readFromStream(stream, callback) {
  var buf = [];
  stream.on("data", onData);
  stream.on("end", onEnd);
  stream.on("error", onError);

  function onData(data) {
    buf.push(data);
  }

  function onEnd() {
    if (callback) callback(null, buf.join(""));
    callback = null;
  }

  function onError(err) {
    if (callback) callback(err);
    callback = null;
  }
}
882431
void getBlockCrc(DataInputStream in, VersionAndOpcode versionAndOpcode)
      throws IOException {
    // header
    BlockChecksumHeader blockChecksumHeader =
        new BlockChecksumHeader(versionAndOpcode);
    blockChecksumHeader.readFields(in);
    final int namespaceId = blockChecksumHeader.getNamespaceId();
    final Block block = new Block(blockChecksumHeader.getBlockId(), 0,
            blockChecksumHeader.getGenStamp());

    DataOutputStream out = null;

    ReplicaToRead ri = datanode.data.getReplicaToRead(namespaceId, block);
    if (ri == null) {
      throw new IOException("Unknown block");
    }

    updateCurrentThreadName("getting CRC checksum for block " + block);

    try {
      //write reply
      out = new DataOutputStream(
          NetUtils.getOutputStream(s, datanode.socketWriteTimeout));
      int blockCrc;
      if (ri.hasBlockCrcInfo()) {
        // There is actually a short window that the block is reopened
        // and we got exception when call getBlockCrc but it's OK. It's
        // only happens for append(). So far we don't optimize for this
        // use case. We can do it later when necessary.
        //
        blockCrc = ri.getBlockCrc();
      } else {
        try {
          if (ri.isInlineChecksum()) {
            blockCrc = BlockInlineChecksumReader.getBlockCrc(datanode, ri,
                namespaceId, block);
          } else {
            blockCrc = BlockWithChecksumFileReader.getBlockCrc(datanode, ri,
                namespaceId, block);
          }
        } catch (IOException ioe) {
          LOG.warn("Exception when getting Block CRC", ioe);
          out.writeShort(DataTransferProtocol.OP_STATUS_ERROR);
          out.flush();
          throw ioe;
        }
      }
      
      out.writeShort(DataTransferProtocol.OP_STATUS_SUCCESS);
      out.writeLong(blockCrc);
      out.flush();
    } finally {
      IOUtils.closeStream(out);
    }
  }
998858
function makeStreamParser() {
  var chunkNb = 0;
  var isBinary = false;

  var faceOffset = 0;
  var previousRemainderData = void 0;

  var parser = function parser(chunk, enc, callback) {
    // console.log('chunk', chunk.length, chunkNb)
    if (chunkNb === 0) {
      isBinary = (0, _utils.isDataBinaryRobust)(chunk);
    }

    var workChunk = void 0;
    if (previousRemainderData) {
      workChunk = new Uint8Array(previousRemainderData.byteLength + chunk.byteLength);
      workChunk.set(new Uint8Array(previousRemainderData));
      workChunk.set(new Uint8Array(chunk), previousRemainderData.byteLength);
      workChunk = workChunk.buffer;
      //let data = ensureString(workChunk)
    } else {
      workChunk = chunk;
    }

    var _ref = isBinary ? computeBinaryOffset(workChunk, chunkNb) : computASCIIOffset(workChunk, chunkNb),
        remainingDataInChunk = _ref.remainingDataInChunk,
        startOffset = _ref.startOffset;

    var parsed = isBinary ? parseBinaryChunk(faceOffset, remainingDataInChunk, startOffset, workChunk) : parseASCIIChunk(workChunk);
    // console.log('faceOffset', faceOffset, 'facesInChunk', facesInChunk, 'remainderDataLength', remainderDataLength, 'current face ', face)

    // update size
    chunkNb += 1;

    previousRemainderData = parsed.remainderData;
    faceOffset = parsed.faceOffset;

    callback(null, parsed.data);
  };

  return parser;
}
737912
function fetchStream() {
	var options = arguments.length <= 0 || arguments[0] === undefined ? {} : arguments[0];
	var callback = arguments[1];

	var cb = callback;
	var stream = null;
	if (cb === undefined) {
		stream = makeStream();
		cb = stream.handler;
	}
	var url = typeof options === 'string' ? options : options.url || options.path;
	if (supportFetch) {
		// TODO support Request object?
		var init = (typeof options === 'undefined' ? 'undefined' : _typeof(options)) === 'object' ? options : {};
		fetch(url, init).then(function (res) {
			if (res.status >= 200 && res.status < 300) {
				pump(res.body.getReader(), (0, _parser2.default)(cb));
			} else {
				// TODO read custom error payload
				cb(null, { status: res.status, statusText: res.statusText });
			}
		}, function (err) {
			cb(null, err);
		});
	} else {
		(function () {
			var parser = (0, _parser2.default)(cb, _parser.BUFFER);
			var opts = (typeof options === 'undefined' ? 'undefined' : _typeof(options)) === 'object' ? _extends({}, options) : {};
			opts.path = url;
			var req = _streamHttp2.default.get(opts, function (res) {
				var status = res.status || res.statusCode;
				if (!(status >= 200 && status < 300)) {
					// TODO read custom error payload
					cb(null, { status: status, statusText: res.statusText || res.statusMessage });
					return;
				}

				res.on('data', function (buf) {
					if (parser(buf) === false) {
						// cancelling
						req.abort();
					}
				});

				res.on('error', function (err) {
					req.abort();
					cb(null, err);
				});
			});
		})();
	}
	return stream;
}
824710
public boolean writeToStream(SendfileOutputStream os, 
                               long offset, long length,
                               long []blockAddresses, long blockLength)
    throws IOException
  {
    throw new UnsupportedOperationException(getClass().getName());
  }