Select input streams.
inProgressOk should be true only for tailing, not for startup
516449
@Override
  public void selectInputStreams(Collection<EditLogInputStream> streams,
      long fromTxnId, boolean inProgressOk, boolean validateInProgressSegments)
      throws IOException {

    QuorumCall<AsyncLogger, RemoteEditLogManifest> q =
        loggers.getEditLogManifest(fromTxnId);
    // we insist on getting all responses, even if they are to be exceptions
    // this will fail if we cannot get majority of successes
    Map<AsyncLogger, RemoteEditLogManifest> resps = loggers
        .waitForReadQuorumWithAllResponses(q, selectInputStreamsTimeoutMs,
            "selectInputStreams");
    
    if(LOG.isDebugEnabled()) {
      LOG.debug("selectInputStream manifests:\n" +
          Joiner.on("\n").withKeyValueSeparator(": ").join(resps));
    }
    
    final PriorityQueue<EditLogInputStream> allStreams = 
        new PriorityQueue<EditLogInputStream>(64,
            JournalSet.EDIT_LOG_INPUT_STREAM_COMPARATOR);
    for (Map.Entry<AsyncLogger, RemoteEditLogManifest> e : resps.entrySet()) {
      AsyncLogger logger = e.getKey();
      RemoteEditLogManifest manifest = e.getValue();
      
      for (RemoteEditLog remoteLog : manifest.getLogs()) {
        EditLogInputStream elis = new URLLogInputStream(logger,
            remoteLog.getStartTxId(), httpConnectReadTimeoutMs);
        if (elis.isInProgress() && !inProgressOk) {
          continue;
        }
        allStreams.add(elis);
      }
    }
    // we pass 0 as min redundance as we do not care about this here
    JournalSet.chainAndMakeRedundantStreams(
        streams, allStreams, fromTxnId, inProgressOk, 0);
  }
1187392
public boolean track(GlobalStreamId stream, long ts) {
        Long currentVal = streamToTs.get(stream);
        if (currentVal == null || ts > currentVal) {
            streamToTs.put(stream, ts);
        }
        checkFailures();
        return ts >= lastWaterMarkTs;
    }
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);
		}
	}
}
593136
def stream_index(self, bucket, index, startkey, endkey=None,
                     return_terms=None, max_results=None, continuation=None,
                     timeout=None):
        
        raise NotImplementedError
1277263
public void init(InputStream in) throws IOException 
    {
        if (TraceComponent.isAnyTracingEnabled()&&logger.isLoggable (Level.FINE)) {  //306998.15
            logger.logp(Level.FINE, CLASS_NAME,"init", "init"); 
        } 
        this.in = in;
        next();
        obs=null;
    }
1308466
public static void loopFor(int start , int endExclusive , IntConsumer consumer ) {
		try {
			pool.submit(() ->IntStream.range(start, endExclusive).parallel().forEach(consumer)).get();
		} catch (InterruptedException | ExecutionException e) {
			e.printStackTrace();
		}
	}
312976
def default_stream_factory(
    total_content_length, filename, content_type, content_length=None
):
    """"""
    max_size = 1024 * 500
    if SpooledTemporaryFile is not None:
        return SpooledTemporaryFile(max_size=max_size, mode="wb+")
    if total_content_length is None or total_content_length > max_size:
        return TemporaryFile("wb+")
    return BytesIO()
204836
def ahead(self, i, j=None):
        '''
        '''
        if j is None:
            return self._stream[self.i + i]
        else:
            return self._stream[self.i + i: self.i + j]
1153253
def _process_streams(cls, streams):
        
        parameterizeds = defaultdict(set)
        valid, invalid = [], []
        for s in streams:
            if isinstance(s, Stream):
                pass
            elif isinstance(s, param.Parameter):
                s = Params(s.owner, [s.name])
            elif isinstance(s, param.Parameterized):
                s = Params(s)
            elif util.is_param_method(s):
                if not hasattr(s, "_dinfo"):
                    continue
                s = ParamMethod(s)
            else:
                invalid.append(s)
                continue
            if isinstance(s, Params):
                pid = id(s.parameterized)
                overlap = (set(s.parameters) & parameterizeds[pid])
                if overlap:
                    pname = type(s.parameterized).__name__
                    param.main.param.warning(
                        'The %s parameter(s) on the %s object have '
                        'already been supplied in another stream. '
                        'Ensure that the supplied streams only specify '
                        'each parameter once, otherwise multiple '
                        'events will be triggered when the parameter '
                        'changes.' % (sorted(overlap), pname))
                parameterizeds[pid] |= set(s.parameters)
            valid.append(s)
        return valid, invalid
595380
public void setProcessInputStream(OutputStream os) {
    if (input != null) {
      if (input == System.in) {
        inputThread = createSystemInPump(input, os);
      }
      else {
        inputThread = createPump(input, os, true, true);
      }
    }
    else {
      try {
        os.close();
      }
      catch (IOException e) {
        log.info("Got exception while closing output stream", e);
      }
    }
  }
1290705
private static Object readDynamic(StreamingInput input)
		throws IOException
	{
		switch(input.peek())
		{
			case VALUE:
				input.next();
				return input.getValue();
			case NULL:
				input.next();
				return input.getValue();
			case LIST_START:
				return readList(input);
			case OBJECT_START:
				return readMap(input);
		}

		throw new SerializationException("Unable to read file, unknown start of value: " + input.peek());
	}
512437
public void print(PrintStream stream) {
    if (mSeries.isEmpty()) {
      return;
    }
    long start = mSeries.firstKey();
    stream.printf("Time series starts at %d with width %d.%n", start, mWidthNano);
    int bucketIndex = 0;
    Iterator<Map.Entry<Long, Integer>> it = mSeries.entrySet().iterator();

    Map.Entry<Long, Integer> current = it.next();
    while (current != null) {
      int numEvents = 0;
      if (bucketIndex * mWidthNano + start == current.getKey()) {
        numEvents = current.getValue();
        current = null;
        if (it.hasNext()) {
          current = it.next();
        }
      }
      stream.printf("%d %d%n", bucketIndex, numEvents);
      bucketIndex++;
    }
  }
305294
protected void skipStatements() throws IOException {
		while (true) {
			st.nextToken();
			if (null != st.sval && st.sval.equalsIgnoreCase("VALUES")) {
				// the next token is the begin of a value
				break;
			}
			if (st.ttype == StreamTokenizer.TT_EOF) {
				// the end of the file is reached
				EOF_reached = true;
				break;
			}
		}
	}
1206430
private ImmutableSet<Pipeline> selectPipelines(InterpreterListener interpreterListener,
                                                   Set<Tuple2<String, String>> processingBlacklist,
                                                   Message message,
                                                   Set<String> initialStreamIds,
                                                   ImmutableSetMultimap<String, Pipeline> streamConnection) {
        final String msgId = message.getId();

        // if a message-stream combination has already been processed (is in the set), skip that execution
        final Set<String> streamsIds = initialStreamIds.stream()
                .filter(streamId -> !processingBlacklist.contains(tuple(msgId, streamId)))
                .filter(streamConnection::containsKey)
                .collect(Collectors.toSet());
        final ImmutableSet<Pipeline> pipelinesToRun = streamsIds.stream()
                .flatMap(streamId -> streamConnection.get(streamId).stream())
                .collect(ImmutableSet.toImmutableSet());
        interpreterListener.processStreams(message, pipelinesToRun, streamsIds);
        log.debug("[{}] running pipelines {} for streams {}", msgId, pipelinesToRun, streamsIds);
        return pipelinesToRun;
    }
1270664
public static void skip(final DataInput in) throws IOException {
		final int length = in.readInt();
		skipFully(in, length);
	}
1282018
def _reduce_input(self, inputs, reducer, final=NotImplemented):
        
        for key, values in groupby(inputs, key=lambda x: self.internal_serialize(x[0])):
            for output in reducer(self.deserialize(key), (v[1] for v in values)):
                yield output
        if final != NotImplemented:
            for output in final():
                yield output
        self._flush_batch_incr_counter()
1070593
private long processNackWithReturnValue(ControlNack nackMsg)
                    throws SIResourceException
    {
        // This is called by a PubSubOutputHandler when it finds Unknown
        // ticks in it's own stream and need the InputStream to resend them
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "processNackWithReturnValue", nackMsg);

        long returnValue = -1;

        // The upstream cellule for this stream is stored in the originStreamMap
        // unless we were the originator.
        SIBUuid12 stream = nackMsg.getGuaranteedStreamUUID();

        if (_sourceStreamManager.hasStream(stream))
        {
            // 242139: Currently we get back to here for every processNack
            // that the InternalOutputStream handles. This is an error but
            // a harmles one. For now the fix is to avoid throwing an exception 
            // as the InternalOutputstream has in fact satisfied the Nack correctly.  
            // The longer term fix is to writeCombined range into the IOS not just the
            // value

            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(tc, "Ignoring processNack on sourceStream at PubSubInputHandler");

            returnValue = _sourceStreamManager.getStreamSet().getStream(nackMsg.getPriority(), nackMsg.getReliability()).getCompletedPrefix();
        }
        else
        {
            // Else we are an IME
            //  send nacks if necessary
            _internalInputStreamManager.processNack(nackMsg);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "processNackWithReturnValue", new Long(returnValue));

        return returnValue;

    }