<commit_before>/**
 * Copyright 2015 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package ch.petikoch.examples.rxjava.parallel;

import rx.Observable;
import rx.Single;
import rx.schedulers.Schedulers;

import java.util.concurrent.ThreadLocalRandom;

public class ParallelExample {

    // https://github.com/ReactiveX/RxJava/issues/1673
    public static void main(String[] args) throws InterruptedException {

        Observable<Integer> numberSource = Observable.range(1, 5)
                .doOnNext(integer1 -> print("Generated: " + integer1));

        Observable<Integer> mergedResults = numberSource.flatMap(
                aNumber -> createSlowlyBlockingPrintObservable(aNumber).subscribeOn(Schedulers.io())
        );

        mergedResults.observeOn(Schedulers.io()).subscribe(aNumber -> print("Finished: " + aNumber));

        // to keep the main thread alive and avoid JVM shutdown
        Thread.sleep(10000);
    }

    private static Observable<Integer> createSlowlyBlockingPrintObservable(Integer item) {
        return Single.<Integer>create(subscriber -> {
            printSlow("Printing " + item + " slow.............. ");
            print("..............ly: " + item + " printed.");
            if (!subscriber.isUnsubscribed()) {
                subscriber.onSuccess(item);
            }
        }).toObservable();
    }

    private static void printSlow(String text) {
        System.out.println("[" + Thread.currentThread().getName() + "] " + text);
        try {
            Thread.sleep(ThreadLocalRandom.current().nextInt(2000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void print(String text) {
        System.out.println("[" + Thread.currentThread().getName() + "] " + text);
    }
}<commit_msg>simplified<commit_after>/**
 * Copyright 2015 <NAME>
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package ch.petikoch.examples.rxjava.parallel;

import rx.Observable;
import rx.Single;
import rx.schedulers.Schedulers;

import java.util.concurrent.ThreadLocalRandom;

public class ParallelExample {

    // https://github.com/ReactiveX/RxJava/issues/1673
    public static void main(String[] args) throws InterruptedException {

        Observable<Integer> numberSource = Observable.range(1, 5)
                .doOnNext(integer1 -> print("Generated: " + integer1));

        Observable<Integer> mergedResults = numberSource.flatMap(
                aNumber -> createSlowlyBlockingPrintObservable(aNumber).subscribeOn(Schedulers.io())
        );

        mergedResults.observeOn(Schedulers.io())
                .toBlocking()
                .forEach(aNumber -> print("Finished: " + aNumber));
    }

    private static Observable<Integer> createSlowlyBlockingPrintObservable(Integer item) {
        return Single.<Integer>create(subscriber -> {
            printSlow("Printing " + item + " slow.............. ");
            print("..............ly: " + item + " printed.");
            if (!subscriber.isUnsubscribed()) {
                subscriber.onSuccess(item);
            }
        }).toObservable();
    }

    private static void printSlow(String text) {
        System.out.println("[" + Thread.currentThread().getName() + "] " + text);
        try {
            Thread.sleep(ThreadLocalRandom.current().nextInt(2000));
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static void print(String text) {
        System.out.println("[" + Thread.currentThread().getName() + "] " + text);
    }
}
{'fraction_non_alphanumeric': 0.12483660130718954, 'fraction_numerical': 0.00980392156862745, 'mean_word_length': 2.7569558101472995, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 21, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, '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': '1892032', 'n_tokens_mistral': 1249, 'n_tokens_neox': 1192, 'n_words': 327}