<commit_before>/*
 * Copyright 2012-2013 Continuuity,Inc. All Rights Reserved.
 */
package com.continuuity.data.stream;

import com.continuuity.data.file.FileReader;
import com.continuuity.data.file.PositionReporter;
import com.continuuity.data2.transaction.stream.StreamConfig;
import com.google.common.base.Function;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.primitives.Longs;
import it.unimi.dsi.fastutil.PriorityQueue;
import it.unimi.dsi.fastutil.objects.ObjectHeapPriorityQueue;

import java.io.Closeable;
import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.NotThreadSafe;

/**
 * A {@link FileReader} that combines multiple event stream into single event stream.
 */
@NotThreadSafe
public final class MultiLiveStreamFileReader implements FileReader<StreamEventOffset, Iterable<StreamFileOffset>> {

  private final PriorityQueue<StreamEventSource> eventSources;
  private final Set<StreamEventSource> emptySources;
  private final Set<StreamEventSource> allSources;
  private final Iterable<StreamFileOffset> offsetsView;

  public MultiLiveStreamFileReader(StreamConfig streamConfig, Iterable<? extends StreamFileOffset> offsets) {
    this.allSources = Sets.newTreeSet();

    for (StreamFileOffset source : offsets) {
      StreamEventSource eventSource = new StreamEventSource(streamConfig, source);
      allSources.add(eventSource);
    }

    this.eventSources = new ObjectHeapPriorityQueue<StreamEventSource>(allSources.size());
    this.emptySources = Sets.newHashSet(allSources);
    this.offsetsView = Iterables.transform(allSources, new Function<StreamEventSource, StreamFileOffset>() {
      @Override
      public StreamFileOffset apply(StreamEventSource input) {
        return input.getPosition();
      }
    });
  }

  @Override
  public int read(Collection<? super StreamEventOffset> events, int maxEvents,
                  long timeout, TimeUnit unit) throws IOException, InterruptedException {

    int eventsRead = 0;

    Stopwatch stopwatch = new Stopwatch();
    stopwatch.start();

    while (eventsRead < maxEvents && !(emptySources.isEmpty() && eventSources.isEmpty())) {
      if (!emptySources.isEmpty()) {
        prepareEmptySources();
      }
      eventsRead += read(events);

      if (stopwatch.elapsedTime(unit) > timeout) {
        break;
      }
    }

    return (eventsRead == 0 && emptySources.isEmpty() && eventSources.isEmpty()) ? -1 : eventsRead;
  }

  @Override<commit_msg>Tries to read as many events as they are available up to batch size, regardless of the timeout value.<commit_after>/*
 * Copyright 2012-2013 Continuuity,Inc. All Rights Reserved.
 */
package com.continuuity.data.stream;

import com.continuuity.data.file.FileReader;
import com.continuuity.data.file.PositionReporter;
import com.continuuity.data2.transaction.stream.StreamConfig;
import com.google.common.base.Function;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.primitives.Longs;
import it.unimi.dsi.fastutil.PriorityQueue;
import it.unimi.dsi.fastutil.objects.ObjectHeapPriorityQueue;

import java.io.Closeable;
import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.NotThreadSafe;

/**
 * A {@link FileReader} that combines multiple event stream into single event stream.
 */
@NotThreadSafe
public final class MultiLiveStreamFileReader implements FileReader<StreamEventOffset, Iterable<StreamFileOffset>> {

  private final PriorityQueue<StreamEventSource> eventSources;
  private final Set<StreamEventSource> emptySources;
  private final Set<StreamEventSource> allSources;
  private final Iterable<StreamFileOffset> offsetsView;

  public MultiLiveStreamFileReader(StreamConfig streamConfig, Iterable<? extends StreamFileOffset> offsets) {
    this.allSources = Sets.newTreeSet();

    for (StreamFileOffset source : offsets) {
      StreamEventSource eventSource = new StreamEventSource(streamConfig, source);
      allSources.add(eventSource);
    }

    this.eventSources = new ObjectHeapPriorityQueue<StreamEventSource>(allSources.size());
    this.emptySources = Sets.newHashSet(allSources);
    this.offsetsView = Iterables.transform(allSources, new Function<StreamEventSource, StreamFileOffset>() {
      @Override
      public StreamFileOffset apply(StreamEventSource input) {
        return input.getPosition();
      }
    });
  }

  @Override
  public int read(Collection<? super StreamEventOffset> events, int maxEvents,
                  long timeout, TimeUnit unit) throws IOException, InterruptedException {

    int eventsRead = 0;

    Stopwatch stopwatch = new Stopwatch();
    stopwatch.start();

    while (eventsRead < maxEvents && !(emptySources.isEmpty() && eventSources.isEmpty())) {
      if (!emptySources.isEmpty()) {
        prepareEmptySources();
      }
      eventsRead += read(events);

      if (eventSources.isEmpty() && stopwatch.elapsedTime(unit) > timeout) {
        break;
      }
    }

    return (eventsRead == 0 && emptySources.isEmpty() && eventSources.isEmpty()) ? -1 : eventsRead;
  }

  @Override
{'fraction_non_alphanumeric': 0.10154565061107117, 'fraction_numerical': 0.004313443565780014, 'mean_word_length': 4.559440559440559, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4336532', 'n_tokens_mistral': 1561, 'n_tokens_neox': 1512, 'n_words': 348}