Top { public static <T, ComparatorT extends Comparator<T> & Serializable> Combine.Globally<T, List<T>> of(int count, ComparatorT compareFn) { return Combine.globally(new TopCombineFn<>(count, compareFn)); } private  Top(); static Combine.Globally<T, List<T>> of(int count, ComparatorT compareFn); static Combine.Globally<T, List<T>> smallest(int count); static Combine.Globally<T, List<T>> largest(int count); static TopCombineFn<T, Natural<T>> largestFn(int count); static TopCombineFn<Long, Natural<Long>> largestLongsFn(int count); static TopCombineFn<Integer, Natural<Integer>> largestIntsFn(int count); static TopCombineFn<Double, Natural<Double>> largestDoublesFn(int count); static TopCombineFn<T, Reversed<T>> smallestFn(int count); static TopCombineFn<Long, Reversed<Long>> smallestLongsFn(int count); static TopCombineFn<Integer, Reversed<Integer>> smallestIntsFn(int count); static TopCombineFn<Double, Reversed<Double>> smallestDoublesFn(int count); static PTransform<PCollection<KV<K, V>>, PCollection<KV<K, List<V>>>> perKey(
          int count, ComparatorT compareFn); static PTransform<PCollection<KV<K, V>>, PCollection<KV<K, List<V>>>> smallestPerKey(int count); static PerKey<K, V, List<V>> largestPerKey(int count);  }
@Test public void testTopEmptyWithIncompatibleWindows() { p.enableAbandonedNodeEnforcement(false); Window<String> windowingFn = Window.into(FixedWindows.of(Duration.standardDays(10L))); PCollection<String> input = p.apply(Create.empty(StringUtf8Coder.of())).apply(windowingFn); expectedEx.expect(IllegalStateException.class); expectedEx.expectMessage("Top"); expectedEx.expectMessage("GlobalWindows"); expectedEx.expectMessage("withoutDefaults"); expectedEx.expectMessage("asSingletonView"); input.apply(Top.of(1, new OrderByLength())); }
@Test public void testCountConstraint() { p.enableAbandonedNodeEnforcement(false); PCollection<String> input = p.apply(Create.of(Arrays.asList(COLLECTION)).withCoder(StringUtf8Coder.of())); expectedEx.expect(IllegalArgumentException.class); expectedEx.expectMessage(Matchers.containsString(">= 0")); input.apply(Top.of(-1, new OrderByLength())); }
@Test public void testDisplayData() { Top.Natural<Integer> comparer = new Top.Natural<>(); Combine.Globally<Integer, List<Integer>> top = Top.of(1234, comparer); DisplayData displayData = DisplayData.from(top); assertThat(displayData, hasDisplayItem("count", 1234)); assertThat(displayData, hasDisplayItem("comparer", comparer.getClass())); }