@Test public void test() throws Exception { StreamSupplier<Integer> source = StreamSupplier.of(3, 1, 3, 2, 5, 1, 4, 3, 2); Executor executor = Executors.newSingleThreadExecutor(); StreamSorterStorage<Integer> storage = StreamSorterStorageImpl.create(executor, INT_SERIALIZER, tempFolder.newFolder().toPath()); StreamSorter<Integer, Integer> sorter = StreamSorter.create(storage, Function.identity(), Integer::compareTo, true, 2); StreamConsumerToList<Integer> consumerToList = StreamConsumerToList.create(); await(source.transformWith(sorter) .streamTo(consumerToList.transformWith(randomlySuspending()))); assertEquals(asList(1, 2, 3, 4, 5), consumerToList.getList()); assertEndOfStream(source, consumerToList); assertEndOfStream(sorter); }
public static <K, T> StreamSorter<K, T> create(StreamSorterStorage<T> storage, Function<T, K> keyFunction, Comparator<K> keyComparator, boolean distinct, int itemsInMemorySize) { return new StreamSorter<>(storage, keyFunction, keyComparator, distinct, itemsInMemorySize); }
StreamSorter implements StreamTransformer<T, T> { public static <K, T> StreamSorter<K, T> create(StreamSorterStorage<T> storage, Function<T, K> keyFunction, Comparator<K> keyComparator, boolean distinct, int itemsInMemorySize) { return new StreamSorter<>(storage, keyFunction, keyComparator, distinct, itemsInMemorySize); } }
StreamSorter implements StreamTransformer<T, T> { public static <K, T> StreamSorter<K, T> create(StreamSorterStorage<T> storage, Function<T, K> keyFunction, Comparator<K> keyComparator, boolean distinct, int itemsInMemorySize) { return new StreamSorter<>(storage, keyFunction, keyComparator, distinct, itemsInMemorySize); } private  StreamSorter(StreamSorterStorage<T> storage,
	                     Function<T, K> keyFunction, Comparator<K> keyComparator, boolean distinct,
	                     int itemsInMemory); }
StreamSorter implements StreamTransformer<T, T> { public static <K, T> StreamSorter<K, T> create(StreamSorterStorage<T> storage, Function<T, K> keyFunction, Comparator<K> keyComparator, boolean distinct, int itemsInMemorySize) { return new StreamSorter<>(storage, keyFunction, keyComparator, distinct, itemsInMemorySize); } private  StreamSorter(StreamSorterStorage<T> storage,
	                     Function<T, K> keyFunction, Comparator<K> keyComparator, boolean distinct,
	                     int itemsInMemory); static StreamSorter<K, T> create(StreamSorterStorage<T> storage,
	                                               Function<T, K> keyFunction, Comparator<K> keyComparator, boolean distinct,
	                                               int itemsInMemorySize); @Override StreamConsumer<T> getInput(); @Override StreamSupplier<T> getOutput(); }
StreamSorter implements StreamTransformer<T, T> { public static <K, T> StreamSorter<K, T> create(StreamSorterStorage<T> storage, Function<T, K> keyFunction, Comparator<K> keyComparator, boolean distinct, int itemsInMemorySize) { return new StreamSorter<>(storage, keyFunction, keyComparator, distinct, itemsInMemorySize); } private  StreamSorter(StreamSorterStorage<T> storage,
	                     Function<T, K> keyFunction, Comparator<K> keyComparator, boolean distinct,
	                     int itemsInMemory); static StreamSorter<K, T> create(StreamSorterStorage<T> storage,
	                                               Function<T, K> keyFunction, Comparator<K> keyComparator, boolean distinct,
	                                               int itemsInMemorySize); @Override StreamConsumer<T> getInput(); @Override StreamSupplier<T> getOutput();  }