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();  }
@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)); }
@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); }
@Test public void testErrorOnSupplier() throws IOException { Executor executor = Executors.newSingleThreadExecutor(); ExpectedException exception = new ExpectedException(); StreamSupplier<Integer> source = StreamSupplier.concat( StreamSupplier.of(3, 1, 3, 2), StreamSupplier.closingWithError(exception) ); StreamSorterStorage<Integer> storage = StreamSorterStorageImpl.create(executor, INT_SERIALIZER, tempFolder.newFolder().toPath()); StreamSorter<Integer, Integer> sorter = StreamSorter.create( storage, Function.identity(), Integer::compareTo, true, 10); StreamConsumerToList<Integer> consumerToList = StreamConsumerToList.create(); Throwable e = awaitException(source.transformWith(sorter) .streamTo(consumerToList)); assertSame(exception, e); assertEquals(0, consumerToList.getList().size()); assertClosedWithError(exception, source, consumerToList); assertClosedWithError(exception, sorter); }
@Test public void testErrorsOnStorage() throws IOException { FailingStreamSorterStorageStub<Integer> failingNewPartitionStorage = FailingStreamSorterStorageStub.<Integer>create().withFailNewPartition(); doTestFailingStorage(failingNewPartitionStorage, (streamPromise, sorter, supplier, consumerToList) -> { Throwable exception = awaitException(streamPromise); assertSame(STORAGE_EXCEPTION, exception); assertClosedWithError(STORAGE_EXCEPTION, sorter); assertClosedWithError(STORAGE_EXCEPTION, supplier, consumerToList); assertTrue(consumerToList.getList().isEmpty()); }); FailingStreamSorterStorageStub<Integer> failingWriteStorage = FailingStreamSorterStorageStub.<Integer>create().withFailWrite(); doTestFailingStorage(failingWriteStorage, (streamPromise, sorter, supplier, consumerToList) -> { Throwable exception = awaitException(streamPromise); assertSame(STORAGE_EXCEPTION, exception); assertClosedWithError(STORAGE_EXCEPTION, sorter); assertClosedWithError(STORAGE_EXCEPTION, supplier, consumerToList); assertTrue(consumerToList.getList().isEmpty()); }); FailingStreamSorterStorageStub<Integer> failingReadStorage = FailingStreamSorterStorageStub.<Integer>create().withFailRead(); doTestFailingStorage(failingReadStorage, (streamPromise, sorter, supplier, consumerToList) -> { Throwable exception = awaitException(streamPromise); assertSame(STORAGE_EXCEPTION, exception); assertClosedWithError(STORAGE_EXCEPTION, sorter); assertEndOfStream(supplier); assertClosedWithError(STORAGE_EXCEPTION, consumerToList); assertTrue(consumerToList.getList().isEmpty()); }); FailingStreamSorterStorageStub<Integer> failingCleanup = FailingStreamSorterStorageStub.<Integer>create().withFailCleanup(); doTestFailingStorage(failingCleanup, (streamPromise, sorter, supplier, consumerToList) -> { await(streamPromise); assertEndOfStream(sorter); assertEndOfStream(supplier, consumerToList); assertEquals(asList(1, 2, 3, 4, 5), consumerToList.getList()); }); }