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

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Comparator;
import java.util.concurrent.TimeUnit;

import junit.framework.TestCase;
import junit.framework.TestSuite;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;

import com.github.davidmoten.rx.Strings;
import com.github.davidmoten.rx.operators.OnSubscribeUsing2;
import com.github.davidmoten.rx.operators.OperatorUnsubscribeEagerly;
import com.github.davidmoten.rx.testing.TestingHelper;

public class BigSortTest extends TestCase {

	private static final Logger log = LoggerFactory
			.getLogger(BigSortTest.class);

	public static TestSuite suite() {
		TestingHelper.includeBackpressureRequestOverflowTest = false;
		return TestingHelper
				.function(sorter(2, 2))
				.waitForTerminalEvent(100, TimeUnit.MILLISECONDS)
				.waitForMoreTerminalEvents(100, TimeUnit.MILLISECONDS)
				// test empty
				.name("testSortOfEmptyReturnsEmpty").fromEmpty().expectEmpty()
				.name("testSortAFew").from(3, 2, 1).expect(1, 2, 3)
				.name("testSortAFew").from(3, 2, 1).expect(1, 2, 3)
				// return suite
				.testSuite(BigSortTest.class);
	}

	public void testLarge() {
		final int n = 3;
		// source is n, n-1, .., 0
		Observable<Integer> source = createDescendingRange(n);
		assertEquals(Observable.range(1, n).toList().toBlocking().single(),
				sorter(n - 1, n - 1).call(source).toList().toBlocking()
						.single());
	}

	private Observable<Integer> createDescendingRange(final int n) {
		return Observable.range(0, n).map(new Func1<Integer, Integer>() {
			@Override
			public Integer call(Integer i) {
				return n - i;
			}
		});
	}

	private static final Charset UTF8 = Charset.forName("UTF-8");

	private static Func1<Observable<Integer>, Observable<Integer>> sorter(
			final int maxToSortInMemoryPerThread, final int maxTempResources) {
		return new Func1<Observable<Integer>, Observable<Integer>>() {

			@Override
			public Observable<Integer> call(Observable<Integer> source) {
				Comparator<Integer> comparator = createComparator();
				Func2<Observable<Integer>, File, Observable<File>> writer = createWriter();
				Func1<File, Observable<Integer>> reader = createReader();
				Func0<File> resourceFactory = createResourceFactory();
				Action1<File> resourceDisposer = createResourceDisposer();

				return BigSort.sort(source, comparator, writer, reader,
						resourceFactory, resourceDisposer,
						maxToSortInMemoryPerThread, maxTempResources,
						Schedulers.immediate());
			}

		};
	}

	private static Comparator<Integer> createComparator() {
		return new Comparator<Integer>() {
			@Override
			public int compare(Integer o1, Integer o2) {
				return o1.compareTo(o2);
			}
		};
	}

	private static Func1<File, Observable<Integer>> createReader() {
		return new Func1<File, Observable<Integer>>() {

			@Override
			public Observable<Integer> call(final File file) {
				Observable<String> strings =
				// read the strings from a file
				Strings.from(file)
				// close the file eagerly
						.lift(OperatorUnsubscribeEagerly.<String> instance());

				return
				// split/join the strings by new line character
				Strings.split(strings, "\n")
				// non-blank lines only
						.filter(nonEmptyLines())
						// log<commit_msg>test with a bigger range<commit_after>package com.github.davidmoten.bigsort;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Comparator;
import java.util.concurrent.TimeUnit;

import junit.framework.TestCase;
import junit.framework.TestSuite;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;

import com.github.davidmoten.rx.Strings;
import com.github.davidmoten.rx.operators.OnSubscribeUsing2;
import com.github.davidmoten.rx.operators.OperatorUnsubscribeEagerly;
import com.github.davidmoten.rx.testing.TestingHelper;

public class BigSortTest extends TestCase {

	private static final Logger log = LoggerFactory
			.getLogger(BigSortTest.class);

	public static TestSuite suite() {
		TestingHelper.includeBackpressureRequestOverflowTest = false;
		return TestingHelper
				.function(sorter(2, 2))
				.waitForTerminalEvent(100, TimeUnit.MILLISECONDS)
				.waitForMoreTerminalEvents(100, TimeUnit.MILLISECONDS)
				// test empty
				.name("testSortOfEmptyReturnsEmpty").fromEmpty().expectEmpty()
				.name("testSortAFew").from(3, 2, 1).expect(1, 2, 3)
				.name("testSortAFew").from(3, 2, 1).expect(1, 2, 3)
				// return suite
				.testSuite(BigSortTest.class);
	}

	public void testLarge() {
		final int n = 100;
		// source is n, n-1, .., 0
		Observable<Integer> source = createDescendingRange(n);
		assertEquals(Observable.range(1, n).toList().toBlocking().single(),
				sorter(2, 2).call(source).toList().toBlocking().single());
	}

	private Observable<Integer> createDescendingRange(final int n) {
		return Observable.range(0, n).map(new Func1<Integer, Integer>() {
			@Override
			public Integer call(Integer i) {
				return n - i;
			}
		});
	}

	private static final Charset UTF8 = Charset.forName("UTF-8");

	private static Func1<Observable<Integer>, Observable<Integer>> sorter(
			final int maxToSortInMemoryPerThread, final int maxTempResources) {
		return new Func1<Observable<Integer>, Observable<Integer>>() {

			@Override
			public Observable<Integer> call(Observable<Integer> source) {
				Comparator<Integer> comparator = createComparator();
				Func2<Observable<Integer>, File, Observable<File>> writer = createWriter();
				Func1<File, Observable<Integer>> reader = createReader();
				Func0<File> resourceFactory = createResourceFactory();
				Action1<File> resourceDisposer = createResourceDisposer();

				return BigSort.sort(source, comparator, writer, reader,
						resourceFactory, resourceDisposer,
						maxToSortInMemoryPerThread, maxTempResources,
						Schedulers.immediate());
			}

		};
	}

	private static Comparator<Integer> createComparator() {

		return new Comparator<Integer>() {
			@Override
			public int compare(Integer o1, Integer o2) {
				return o1.compareTo(o2);
			}
		};
	}

	private static Func1<File, Observable<Integer>> createReader() {
		return new Func1<File, Observable<Integer>>() {

			@Override
			public Observable<Integer> call(final File file) {
				Observable<String> strings =
				// read the strings from a file
				Strings.from(file)
				// close the file eagerly
						.lift(OperatorUnsubscribeEagerly.<String> instance());

				return
				// split/join the strings by new line character
				Strings.split(strings, "\n")
				// non-blank lines only
						.filter(nonEmptyLines())
						// log
{'fraction_non_alphanumeric': 0.11852267939896081, 'fraction_numerical': 0.01404297149276787, 'mean_word_length': 4.482678983833718, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 59, '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': '3313756', 'n_tokens_mistral': 2559, 'n_tokens_neox': 2195, 'n_words': 482}