Reads the log file and calls back {@link RedoLog.ActionCollector}.

@param collector called back for each {@link MultiIndex.Action} read.
@throws IOException if an error occurs while reading from the
log file.
331909
private void read(final ActionCollector collector) throws IOException
   {
      if (!dir.fileExists(REDO_LOG))
      {
         return;
      }
      InputStream in = new IndexInputStream(dir.openInput(REDO_LOG));
      BufferedReader reader = null;
      try
      {
         reader = new BufferedReader(new InputStreamReader(in));
         String line;
         while ((line = reader.readLine()) != null)
         {
            try
            {
               collector.collect(MultiIndex.Action.fromString(line));
            }
            catch (IllegalArgumentException e)
            {
               log.warn("Malformed redo entry: " + e.getMessage());
            }
         }
      }
      finally
      {
         if (reader != null)
         {
            try
            {
               reader.close();
            }
            catch (IOException e)
            {
               log.warn("Exception while closing redo log: " + e.toString());
            }
         }
         if (in != null)
         {
            try
            {
               in.close();
            }
            catch (IOException e)
            {
               log.warn("Exception while closing redo log: " + e.toString());
            }
         }
      }
   }
492997
@Override
    public void send(final File file, final CallbackHandler handler)
    {
        log.info("Sending local file to collector: {}", file.getAbsolutePath());
        final long startTime = System.nanoTime();

        final AsyncCompletionHandler<Response> asyncCompletionHandler = new AsyncCompletionHandler<Response>()
        {
            @Override
            public Response onCompleted(final Response response)
            {
                activeRequests.decrementAndGet();

                if (response.getStatusCode() == 202) {
                    handler.onSuccess(file);
                }
                else {
                    handler.onError(new IOException(String.format("Received response %d: %s",
                                                                  response.getStatusCode(), response.getStatusText())), file);
                }

                sendTimer.update(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
                return response; // never read
            }

            @Override
            public void onThrowable(final Throwable t)
            {
                activeRequests.decrementAndGet();
                handler.onError(t, file);
            }
        };

        final HttpJob job = new HttpJob(client, file, asyncCompletionHandler);
        workers.offer(job);
    }
1194719
@Override
    public <R, A> R collect(Collector<? super T, A, R> collector) {
        Predicate<A> finished = finished(collector);
        if (finished != null) {
            BiConsumer<A, ? super T> acc = collector.accumulator();
            BinaryOperator<A> combiner = collector.combiner();
            Spliterator<T> spliterator = spliterator();
            if (!isParallel()) {
                A a = collector.supplier().get();
                if (!finished.test(a)) {
                    try {
                        // forEachRemaining can be much faster
                        // and take much less memory than tryAdvance for certain
                        // spliterators
                        spliterator.forEachRemaining(e -> {
                            acc.accept(a, e);
                            if (finished.test(a))
                                throw new CancelException();
                        });
                    } catch (CancelException ex) {
                        // ignore
                    }
                }
                return collector.finisher().apply(a);
            }
            Spliterator<A> spltr;
            if (!spliterator.hasCharacteristics(Spliterator.ORDERED)
                || collector.characteristics().contains(Characteristics.UNORDERED)) {
                spltr = new UnorderedCancellableSpliterator<>(spliterator, collector.supplier(), acc, combiner,
                        finished);
            } else {
                spltr = new OrderedCancellableSpliterator<>(spliterator, collector.supplier(), acc, combiner, finished);
            }
            return collector.finisher().apply(
                new StreamEx<>(StreamSupport.stream(spltr, true), context).findFirst().get());
        }
        return rawCollect(collector);
    }
229425
private static void reduceItems(FileOutputCollector collector, Reducer reducable, ArrayList<String> items)
            throws IOException {
        if(items != null) {
            // call reducer
            reducable.reduce(items, collector);
        }
    }
325240
public BioCCollection readCollection()
      throws XMLStreamException {
    if (collection != null) {
      BioCCollection thisCollection = collection;
      collection = null;
      return thisCollection;
    }
    return null;
  }
491613
public void push(Collector collector, String job) throws IOException {
    CollectorRegistry registry = new CollectorRegistry();
    collector.register(registry);
    push(registry, job);
  }
1151910
public void timeCollectionsStreaming(int reps) throws IOException {
    for (int i=0; i<reps; ++i) {
      StringReader reader = new StringReader(json);
      JsonReader jr = new JsonReader(reader);
      jr.beginArray();
      List<BagOfPrimitives> bags = new ArrayList<BagOfPrimitives>();
      while(jr.hasNext()) {
        jr.beginObject();
        long longValue = 0;
        int intValue = 0;
        boolean booleanValue = false;
        String stringValue = null;
        while(jr.hasNext()) {
          String name = jr.nextName();
          if (name.equals("longValue")) {
            longValue = jr.nextLong();
          } else if (name.equals("intValue")) {
            intValue = jr.nextInt();
          } else if (name.equals("booleanValue")) {
            booleanValue = jr.nextBoolean();
          } else if (name.equals("stringValue")) {
            stringValue = jr.nextString();
          } else {
            throw new IOException("Unexpected name: " + name);
          }
        }
        jr.endObject();
        bags.add(new BagOfPrimitives(longValue, intValue, booleanValue, stringValue));
      }
      jr.endArray();
    }
  }
656817
public static Iterable<BedRecord> read(final Readable readable) throws IOException {
        checkNotNull(readable);
        Collect collect = new Collect();
        stream(readable, collect);
        return collect.records();
    }
1269509
protected T read(JsonReader in, TypeAdapter<JsonElement> elementAdapter, TypeAdapter<T> delegate) throws IOException {
        JsonElement tree = elementAdapter.read(in);
        afterRead(tree);
        return delegate.fromJsonTree(tree);
    }
1086235
public synchronized static void read(int fd, Collection<ByteBuffer> collection) throws IOException{
        collection.add(ByteBuffer.wrap(read(fd)));
    }
1090753
private void run() throws IOException
   {
      List<MultiIndex.Action> actions = redoLog.getActions();

      // find loser transactions
      for (Iterator<MultiIndex.Action> it = actions.iterator(); it.hasNext();)
      {
         MultiIndex.Action a = it.next();
         if (a.getType() == MultiIndex.Action.TYPE_START)
         {
            losers.add(new Long(a.getTransactionId()));
         }
         else if (a.getType() == MultiIndex.Action.TYPE_COMMIT)
         {
            losers.remove(new Long(a.getTransactionId()));
         }
      }

      // find last volatile commit without changes from a loser
      int lastSafeVolatileCommit = -1;
      Set<Long> transactionIds = new HashSet<Long>();
      for (int i = 0; i < actions.size(); i++)
      {
         MultiIndex.Action a = (MultiIndex.Action)actions.get(i);
         if (a.getType() == MultiIndex.Action.TYPE_COMMIT)
         {
            transactionIds.clear();
         }
         else if (a.getType() == MultiIndex.Action.TYPE_VOLATILE_COMMIT)
         {
            transactionIds.retainAll(losers);
            // check if transactionIds contains losers
            if (transactionIds.size() > 0)
            {
               // found dirty volatile commit
               break;
            }
            else
            {
               lastSafeVolatileCommit = i;
            }
         }
         else
         {
            transactionIds.add(new Long(a.getTransactionId()));
         }
      }

      // delete dirty indexes
      for (int i = lastSafeVolatileCommit + 1; i < actions.size(); i++)
      {
         MultiIndex.Action a = (MultiIndex.Action)actions.get(i);
         if (a.getType() == MultiIndex.Action.TYPE_CREATE_INDEX)
         {
            a.undo(index);
         }
      }

      // replay actions up to last safe volatile commit
      // ignore add node actions, they are included in volatile commits
      for (int i = 0; i < actions.size() && i <= lastSafeVolatileCommit; i++)
      {
         MultiIndex.Action a = (MultiIndex.Action)actions.get(i);
         switch (a.getType())
         {
            case MultiIndex.Action.TYPE_ADD_INDEX :
            case MultiIndex.Action.TYPE_CREATE_INDEX :
            case MultiIndex.Action.TYPE_DELETE_INDEX :
            case MultiIndex.Action.TYPE_DELETE_NODE :
               // ignore actions by the index merger.
               // the previously created index of a merge has been
               // deleted because it was considered dirty.
               // we are conservative here and let the index merger do
               // its work again.
               if (a.getTransactionId() == MultiIndex.Action.INTERNAL_TRANS_REPL_INDEXES)
               {
                  continue;
               }
               a.execute(index);
         }
      }

      // now replay the rest until we encounter a loser transaction
      for (int i = lastSafeVolatileCommit + 1; i < actions.size(); i++)
      {
         MultiIndex.Action a = (MultiIndex.Action)actions.get(i);
         if (losers.contains(new Long(a.getTransactionId())))
         {
            break;
         }
         else
         {
            // ignore actions by the index merger.
            if (a.getTransactionId() == MultiIndex.Action.INTERNAL_TRANS_REPL_INDEXES)
            {
               continue;
            }
            a.execute(index);
         }
      }

      // now we are consistent again -> flush
      index.flush();
      index.releaseMultiReader();
   }
133284
@Override
    public void readExternal(PofReader reader) throws IOException {
        factory  = (Factory<List<E>>) reader.readObject(0);
        delegate = factory.create();
        reader.readCollection(1, delegate);
    }
511851
@Override
    public void stateChanged(ChangeEvent e) {
        log.debug("getting new collector");
        collector = (CorrectedResultCollector) createTestElement();
        if (collector instanceof CorrectedResultCollector) {
            setUpFiltering((CorrectedResultCollector) collector);
        }
        collector.loadExistingFile();
    }
1011012
@Override
  public int read() throws IOException {
    if (mPosition == mLength) { // at end of file
      return -1;
    }
    CountingRetry retry = new CountingRetry(mBlockWorkerClientReadRetry);
    IOException lastException = null;
    while (retry.attempt()) {
      try {
        updateStream();
        int result = mBlockInStream.read();
        if (result != -1) {
          mPosition++;
        }
        return result;
      } catch (UnavailableException | DeadlineExceededException | ConnectException e) {
        lastException = e;
        if (mBlockInStream != null) {
          handleRetryableException(mBlockInStream, e);
          mBlockInStream = null;
        }
      }
    }
    throw lastException;
  }
872019
public void read(BufferedReader reader) throws Exception {
		String line;

		clear();

		while ((line = reader.readLine()) != null) {
			line = line.trim();
			// comment?
			if (line.startsWith("#"))
				continue;
			add(line);
		}

		reader.close();
	}
485149
private void read() throws IOException {
        lastModified = jobFile.lastModified();
        if (jobFile.exists()) {
            try (final InputStream in = new FileInputStream(jobFile)) {
                final XMLReader parser = XMLUtils.getXMLReader();
                parser.setContentHandler(new JobHandler(prop, files));

                parser.parse(new InputSource(in));
            } catch (final SAXException e) {
                throw new IOException("Failed to read job file: " + e.getMessage());
            }
        } else {
            // defaults
            prop.put(PROPERTY_GENERATE_COPY_OUTER, Generate.NOT_GENERATEOUTTER.toString());
            prop.put(PROPERTY_ONLY_TOPIC_IN_MAP, Boolean.toString(false));
            prop.put(PROPERTY_OUTER_CONTROL, OutterControl.WARN.toString());
        }
    }
506441
public static <T extends Collection<String>> T readLines(URL url, Charset charset, T collection) throws IORuntimeException {
		InputStream in = null;
		try {
			in = url.openStream();
			return IoUtil.readLines(in, charset, collection);
		} catch (IOException e) {
			throw new IORuntimeException(e);
		} finally {
			IoUtil.close(in);
		}
	}