@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()); }); }
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();  }