Initialize an actor for the given `Stream`.

@param {Stream} stream
@api public
775190
function Actor(stream) {
  if (!(this instanceof Actor)) return new Actor(stream);
  var that = this;
  this.parser = new amp.Stream;
  this.parser.on('data', this.onmessage.bind(this));
  stream.pipe(this.parser);
  this.stream = stream;
  this.callbacks = {};
  this.ids = 0;
  this.id = ++ids;
  this.secret_key = null;
  Actor.emit('actor', this);
}
346002
function setStreamType(constraints, stream) {
    if (constraints.mandatory && constraints.mandatory.chromeMediaSource) {
        stream.isScreen = true;
    } else if (constraints.mozMediaSource || constraints.mediaSource) {
        stream.isScreen = true;
    } else if (constraints.video) {
        stream.isVideo = true;
    } else if (constraints.audio) {
        stream.isAudio = true;
    }
}
857047
function consumeStream(stream, callback) {
    var chunks = [];
    stream.on("data", onData);
    stream.on("end", onEnd);
    stream.on("error", onError);
    function onData(chunk) {
        chunks.push(chunk);
    }
    function onEnd() {
        cleanup();
        callback(null, chunks.join(""));
    }
    function onError(err) {
        cleanup();
        callback(err);
    }
    function cleanup() {
        stream.removeListener("data", onData);
        stream.removeListener("end", onEnd);
        stream.removeListener("error", onError);
    }
}
960947
protected void init(InputStream inputStream) throws IOException {
		stream = inputStream;
		st = new StreamTokenizer(new BufferedReader(new InputStreamReader(
				stream, ENCODING)));

		EOF_reached = false;
		skipStatements();

	}
1227747
public function initializeStreamWrapper()
    {
        $stream = new Stream('accompli', array(
                'recipe' => realpath(__DIR__.'/Resources/recipe'),
            ), false);

        StreamManager::create()->registerStream($stream);
    }
74781
function init()
  {
    // the stream has now been initialized
    initialized = true;

    // if we have a result
    if (context.result)
    {
      // if no value was specified for the start index or if the specified start
      // index is negative, default to 0, otherwise truncate the fractional part
      start = (!Util.isNumber(start) || (start < 0)) ? 0 : Math.floor(start);

      // if no value was specified for the end index or if the end index is
      // larger than the row index of the last row, default to the index of the
      // last row, otherwise truncate the fractional part
      var returnedRows = context.result.getReturnedRows();
      end = (!Util.isNumber(end) || (end >= returnedRows)) ? returnedRows - 1 :
          Math.floor(end);

      // find all the chunks that overlap with the specified range
      var overlappingChunks = context.result.findOverlappingChunks(start, end);

      // if no chunks overlap or start is greater than end, we're done
      if ((overlappingChunks.length === 0) || (start > end))
      {
        process.nextTick(close);
      }
      else
      {
        // create a result stream from the overlapping chunks
        resultStream = new ResultStream(
        {
          chunks       : overlappingChunks,
          prefetchSize : context.connectionConfig.getResultPrefetch()
        });

        readNextRow();
      }
    }
    else
    {
      close(context.resultError);
    }
  }
874490
function startStream () {
  process.stdin.setEncoding('utf8')
  process.stdin.on('data', function(data) {
    return {
      "--parse"     : function (string) {
        var result = parse(string)
        process.stdout.write(JSON.stringify(result, null, 2))
      },
      "--serialize" : function (string) {
        var result = serialize(JSON.parse(string))
        process.stdout.write(result)
      }
    }[$mode](data)
  })
}
1161990
function streamToValue (response, callback) {
  pump(
    response,
    concat((data) => callback(null, data)),
    (err) => {
      if (err) {
        callback(err)
      }
    })
}
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 );
		});
	}
119713
public void open(InputStream stream) throws IOException
    {

        int size = (int) (stream.available() / UNIT_SIZE);
        _array = new int[size];

        DataInputStream in = null;
        try
        {
            in = new DataInputStream(new BufferedInputStream(
                    stream));
            for (int i = 0; i < size; ++i)
            {
                _array[i] = in.readInt();
            }
        }
        finally
        {
            if (in != null)
            {
                in.close();
            }
        }
    }
500641
function Parser(stream) {
  StreamStack.call(this, stream, {
    data: function(b) { this._onData(b); }
  });
  this._onData = this._parseHeader;
  this._headerParser = new HeaderParser(new Stream(), {
    emitFirstLine: false,
    strictCRLF: false,
    strictSpaceAfterColon: false,
    allowFoldedHeaders: false
  });
  this._headerParser.on('headers', this._onHeadersComplete.bind(this));
}
110884
function setUpWritable(stream, serialize) {
  stream.finished = false;
  stream.status = {
    code : constants.status.OK,
    details : 'OK',
    metadata : new Metadata()
  };
  stream.serialize = common.wrapIgnoreNull(serialize);
  function sendStatus() {
    var batch = {};
    if (!stream.call.metadataSent) {
      stream.call.metadataSent = true;
      batch[grpc.opType.SEND_INITIAL_METADATA] =
          (new Metadata())._getCoreRepresentation();
    }

    if (stream.status.metadata) {
      stream.status.metadata = stream.status.metadata._getCoreRepresentation();
    }
    batch[grpc.opType.SEND_STATUS_FROM_SERVER] = stream.status;
    stream.call.startBatch(batch, function(){});
  }
  stream.on('finish', sendStatus);
  /**
   * Set the pending status to a given error status. If the error does not have
   * code or details properties, the code will be set to grpc.status.UNKNOWN
   * and the details will be set to 'Unknown Error'.
   * @param {Error} err The error object
   */
  function setStatus(err) {
    var code = constants.status.UNKNOWN;
    var details = 'Unknown Error';
    var metadata = new Metadata();
    if (err.hasOwnProperty('message')) {
      details = err.message;
    }
    if (err.hasOwnProperty('code')) {
      code = err.code;
      if (err.hasOwnProperty('details')) {
        details = err.details;
      }
    }
    if (err.hasOwnProperty('metadata')) {
      metadata = err.metadata;
    }
    stream.status = {code: code, details: details, metadata: metadata};
  }
  /**
   * Terminate the call. This includes indicating that reads are done, draining
   * all pending writes, and sending the given error as a status
   * @param {Error} err The error object
   * @this GrpcServerStream
   */
  function terminateCall(err) {
    // Drain readable data
    setStatus(err);
    stream.end();
  }
  stream.on('error', terminateCall);
  /**
   * Override of Writable#end method that allows for sending metadata with a
   * success status.
   * @param {Metadata=} metadata Metadata to send with the status
   */
  stream.end = function(metadata) {
    if (metadata) {
      stream.status.metadata = metadata;
    }
    Writable.prototype.end.call(this);
  };
}
476682
public void parseStream (InputStream stream)
        throws IOException
    {
        try {
            // read the XML input stream and construct the scene object
            _chars = new StringBuilder();
            XMLUtil.parse(this, stream);

        } catch (ParserConfigurationException pce) {
            throw (IOException) new IOException().initCause(pce);

        } catch (SAXException saxe) {
            throw (IOException) new IOException().initCause(saxe);
        }
    }
992161
function( stream ){
			if( typeof stream == 'string' ){
				//if node.js create a writeable stream with this filename
			} else if( internals.hasProperties(stream,['write','end','writable'] && stream.writable)){
				this.objStream = stream;
			} else {
			}
			this._handleBeginSave();
		}
320281
public static <T, U> Stream<U> cast(final Stream<T> stream, final Class<? extends U> type) {
        return stream.map(type::cast);
    }
849139
function Stream(options) {

  /**
   * Whether or not the Stream was initialized as
   * a single- or multi-key stream.
   *
   * @property multi
   * @type Boolean
   */
  reader(this, 'multi', !(options && options.apiKey));

  if (!this.multi) {
    options.headers = options.headers || {};
    options.headers['Zotero-API-Key'] = options.apiKey;

    delete options.apiKey;
  }

  Client.call(this, options);

  /**
   * @property retry
   * @type Object
   */
  reader(this, 'retry', {
    delay: Stream.DEFAULT_RETRY
  });

  /**
   * @property idle
   * @type Object
   */
  reader(this, 'idle', {
    ping: Stream.IDLE_TIMEOUT,
    pong: Stream.IDLE_TIMEOUT / 2
  });

  /**
   * @property subscriptions
   * @type Subscriptions
   */
  this.subscriptions = new Subscriptions();

  this.open = this.open.bind(this);
  this.opened = this.opened.bind(this);
  this.closed = this.closed.bind(this);
  this.ping = this.ping.bind(this);
  this.pong = this.pong.bind(this);
  this.receive = this.receive.bind(this);
  this.error = this.error.bind(this);

  this.open();
}
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);
            }
        }
    }
}