@Test public void testStreamStorage() { StreamSupplier<Integer> source1 = StreamSupplier.of(1, 2, 3, 4, 5, 6, 7); Executor executor = Executors.newSingleThreadExecutor(); StreamSorterStorageImpl<Integer> storage = StreamSorterStorageImpl.create(executor, INT_SERIALIZER, tempFolder.getRoot().toPath()) .withWriteBlockSize(MemSize.of(64)); StreamConsumer<Integer> writer1 = storage.writeStream(1); source1.streamTo(writer1); Eventloop.getCurrentEventloop().run(); assertEndOfStream(source1); StreamConsumerToList<Integer> consumer1 = StreamConsumerToList.create(); storage.readStream(1).streamTo(consumer1.transformWith(oneByOne())); Eventloop.getCurrentEventloop().run(); assertEquals(asList(1, 2, 3, 4, 5, 6, 7), consumer1.getList()); storage.cleanup(Arrays.asList(1, 2)); }
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();  }