<commit_before>package com.github.davidmoten.bigsort;

import java.io.File;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

import rx.Notification;
import rx.Observable;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.functions.FuncN;

public class BigSort {

	public static <T> Observable<T> sort(Observable<T> source,
			final Comparator<T> comparator,
			final Func2<Observable<T>, File, Observable<File>> writer,
			final Func1<File, Observable<T>> reader,
			final Func0<File> fileFactory, int maxToSortInMemoryPerThread,
			final int maxTempFiles) {
		return source
		// buffer into groups small enough to sort in memory
				.buffer(maxToSortInMemoryPerThread)
				// sort each buffer to a file
				.flatMap(
						sortInMemoryAndWriteToAFile(comparator, writer,
								fileFactory))
				// make each file an Observable<File>
				.nest()
				// reduce by merging the files to a single file once the file
				// count is maxTempFiles
				.reduce(Observable.<File> empty(),
						mergeFiles(comparator, writer, reader, fileFactory,
								maxTempFiles))
				// flatten to a sequence of File
				.flatMap(
						com.github.davidmoten.rx.Functions
								.<Observable<File>> identity())
				// accumulate to a list
				.toList()
				// merge remaining files
				.flatMap(mergeFileList(comparator, reader));
	}

	private static <T> Func1<List<T>, Observable<File>> sortInMemoryAndWriteToAFile(
			final Comparator<T> comparator,
			final Func2<Observable<T>, File, Observable<File>> writer,
			final Func0<File> fileFactory) {
		return new Func1<List<T>, Observable<File>>() {
			@Override
			public Observable<File> call(List<T> list) {
				return Observable.just(list)
				// sort
						.map(sortList(comparator))
						// write to file
						.flatMap(writeToFile(writer, fileFactory));
			}
		};
	}

	private static <T> Func2<Observable<File>, Observable<File>, Observable<File>> mergeFiles(
			final Comparator<T> comparator,
			final Func2<Observable<T>, File, Observable<File>> writer,
			final Func1<File, Observable<T>> reader,
			final Func0<File> fileFactory, final int maxTempFiles) {
		return new Func2<Observable<File>, Observable<File>, Observable<File>>() {

			@Override
			public Observable<File> call(Observable<File> files,
					final Observable<File> f) {
				return files
						.concatWith(f)
						.toList()
						.flatMap(
								mergeWhenSizeIsMaxTempFiles(comparator, writer,
										reader, fileFactory, maxTempFiles));
			}

		};
	}

	private static <T> Func1<List<File>, Observable<T>> mergeFileList(
			final Comparator<T> comparator,
			final Func1<File, Observable<T>> reader) {<commit_msg>cleanup<commit_after>package com.github.davidmoten.bigsort;

import java.io.File;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

import rx.Notification;
import rx.Observable;
import rx.Scheduler;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.functions.FuncN;

public class BigSort {

	public static <T> Observable<T> sort(Observable<T> source,
			final Comparator<T> comparator,
			final Func2<Observable<T>, File, Observable<File>> writer,
			final Func1<File, Observable<T>> reader,
			final Func0<File> fileFactory, int maxToSortInMemoryPerThread,
			final int maxTempFiles, Scheduler scheduler) {
		return source
		// buffer into groups small enough to sort in memory
				.buffer(maxToSortInMemoryPerThread)
				// sort each buffer to a file
				.flatMap(
						sortInMemoryAndWriteToAFile(comparator, writer,
								fileFactory, scheduler))
				// make each file an Observable<File>
				.nest()
				// reduce by merging the files to a single file once the file
				// count is maxTempFiles
				.reduce(Observable.<File> empty(),
						mergeFiles(comparator, writer, reader, fileFactory,
								maxTempFiles))
				// flatten to a sequence of File
				.flatMap(
						com.github.davidmoten.rx.Functions
								.<Observable<File>> identity())
				// accumulate to a list
				.toList()
				// merge remaining files
				.flatMap(mergeFileList(comparator, reader));
	}

	private static <T> Func1<List<T>, Observable<File>> sortInMemoryAndWriteToAFile(
			final Comparator<T> comparator,
			final Func2<Observable<T>, File, Observable<File>> writer,
			final Func0<File> fileFactory, final Scheduler scheduler) {
		return new Func1<List<T>, Observable<File>>() {
			@Override
			public Observable<File> call(List<T> list) {
				return Observable.just(list)
				// sort
						.map(sortList(comparator))
						// write to file
						.flatMap(writeToFile(writer, fileFactory))
						// subscribe on desired scheduler
						.subscribeOn(scheduler);
			}
		};
	}

	private static <T> Func2<Observable<File>, Observable<File>, Observable<File>> mergeFiles(
			final Comparator<T> comparator,
			final Func2<Observable<T>, File, Observable<File>> writer,
			final Func1<File, Observable<T>> reader,
			final Func0<File> fileFactory, final int maxTempFiles) {
		return new Func2<Observable<File>, Observable<File>, Observable<File>>() {

			@Override
			public Observable<File> call(Observable<File> files,
					final Observable<File> f) {
				return files
						.concatWith(f)
						.toList()
						.flatMap(
								mergeWhenSizeIsMaxTempFiles(comparator, writer,
										reader, fileFactory, maxTempFiles));
			}

		};
	}

	private static <T> Func1<List<File>, Observable<T>> mergeFileList(
			final Comparator<T> comparator,
			final Func1<File, Observable<T>> reader) {
{'fraction_non_alphanumeric': 0.11202283267927221, 'fraction_numerical': 0.006064930431680342, 'mean_word_length': 3.8461538461538463, 'pattern_counts': {'":': 0, '<': 111, '<?xml version=': 0, '>': 111, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3518909', 'n_tokens_mistral': 2128, 'n_tokens_neox': 1738, 'n_words': 451}