<commit_before>package se.callista.springmvc.asynch.pattern.aggregator;

import com.ning.http.client.Response;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.task.TaskExecutor;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult;
import rx.Observable;
import rx.Subscription;
import rx.schedulers.Schedulers;
import se.callista.springmvc.asynch.common.lambdasupport.AsyncHttpClientRx;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author <NAME> <<EMAIL>>
 */
@RestController
public class AggregatorNonBlockingRxController {

    private static final Logger LOG = LoggerFactory.getLogger(AggregatorNonBlockingRxController.class);

    @Value("${sp.non_blocking.url}")
    private String SP_NON_BLOCKING_URL;

    @Value("${aggregator.timeoutMs}")
    private int TIMEOUT_MS;

    @Autowired
    @Qualifier("dbThreadPoolExecutor")
    private TaskExecutor dbThreadPoolExecutor;

    @Autowired
    private AsyncHttpClientRx asyncHttpClientRx;

    /**
     * Sample usage: curl "http://localhost:9080/aggregate-non-blocking-rx?minMs=1000&maxMs=2000"
     *
     * @param dbLookupMs time for db lookup
     * @param dbHits number of results from db
     * @param minMs min execution time of remote
     * @param maxMs max execution time of remote
     *
     * @return deferred result
     *
     * @throws java.io.IOException
     */
    @RequestMapping("/aggregate-non-blocking-rx")
    public DeferredResult<String> nonBlockingAggregator(
        @RequestParam(value = "dbLookupMs", required = false, defaultValue = "0")    int dbLookupMs,
        @RequestParam(value = "dbHits",     required = false, defaultValue = "3")    int dbHits,
        @RequestParam(value = "minMs",      required = false, defaultValue = "0")    int minMs,
        @RequestParam(value = "maxMs",      required = false, defaultValue = "0")    int maxMs) throws IOException {

        DbLookup dbLookup = new DbLookup(dbLookupMs, dbHits);
        DeferredResult<String> deferredResult = new DeferredResult<>();

        Subscription subscription =
                Observable.from(dbLookup.lookupUrlsInDb(SP_NON_BLOCKING_URL, minMs, maxMs))
                .subscribeOn(Schedulers.from(dbThreadPoolExecutor))
                .flatMap(u ->
                    asyncHttpClientRx
                        .observable(u)
                        .map(this::getResponseBody)
                        .onErrorReturn(t -> "error")
                )
                .doOnNext(this::logThread)
                .observeOn(Schedulers.computation())
                .doOnNext(this::logThread)
                .buffer(TIMEOUT_MS, TimeUnit.MILLISECONDS, dbHits)
                .subscribe(v -> deferredResult.setResult(getTotalResult(v)));

        deferredResult.onCompletion(subscription::unsubscribe);

        return deferredResult;
    }

    private void logThread(String r) {
        LOG.debug("Thread:[" + Thread.currentThread().getName()+"] : " + r);
    }

    private String getResponseBody(Response response) {
        try {
            return response.getResponseBody();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private String getTotalResult(List<String> resultArr) {
        String totalResult = "";
        for (String r : resultArr)
            totalResult += r + '\n';
        return totalResult;
    }

}<commit_msg>Updated Rx Aggregator to more reflect the Callback example<commit_after>package se.callista.springmvc.asynch.pattern.aggregator;

import com.ning.http.client.Response;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.task.TaskExecutor;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult;
import rx.Observable;
import rx.Subscription;
import rx.schedulers.Schedulers;
import se.callista.springmvc.asynch.common.lambdasupport.AsyncHttpClientRx;

import java.io.IOException;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

/**
 * @author <NAME> <<EMAIL>>
 */
@RestController
public class AggregatorNonBlockingRxController {

    private static final Logger LOG = LoggerFactory.getLogger(AggregatorNonBlockingRxController.class);

    @Value("${sp.non_blocking.url}")
    private String SP_NON_BLOCKING_URL;

    @Value("${aggregator.timeoutMs}")
    private int TIMEOUT_MS;

    @Autowired
    @Qualifier("dbThreadPoolExecutor")
    private TaskExecutor dbThreadPoolExecutor;

    @Autowired
    private AsyncHttpClientRx asyncHttpClientRx;

    private class Result {
        private String url = "";
        private int id;

        Result() {}

        Result(int id, String url) {
            this.id = id;
            this.url = url;
        }

        @Override
        public String toString() {
            return "Result{" +
                    "url='" + url + '\'' +
                    ", index=" + id +
                    '}';
        }
    }

    /**
     * Sample usage: curl "http://localhost:9080/aggregate-non-blocking-rx?minMs=1000&maxMs=2000"
     *
     * @param dbLookupMs time for db lookup
     * @param dbHits number of results from db
     * @param minMs min execution time of remote
     * @param maxMs max execution time of remote
     *
     * @return deferred result
     *
     * @throws java.io.IOException
     */
    @RequestMapping("/aggregate-non-blocking-rx")
    public DeferredResult<String> nonBlockingAggregator(
        @RequestParam(value = "dbLookupMs", required = false, defaultValue = "0")    int dbLookupMs,
        @RequestParam(value = "dbHits",     required = false, defaultValue = "3")    int dbHits,
        @RequestParam(value = "minMs",      required = false, defaultValue = "0")    int minMs,
        @RequestParam(value = "maxMs",      required = false, defaultValue = "0")    int maxMs) throws IOException {

        DbLookup dbLookup = new DbLookup(dbLookupMs, dbHits);
        DeferredResult<String> deferredResult = new DeferredResult<>();

        Subscription subscription =
                Observable.from(dbLookup.lookupUrlsInDb(SP_NON_BLOCKING_URL, minMs, maxMs))
                .subscribeOn(Schedulers.from(dbThreadPoolExecutor))
                .scan(new Result(), (result, url) -> new Result(result.id + 1, url))
                .filter(result -> result.id > 0)
                .flatMap(result ->
                    asyncHttpClientRx
                        .observable(result.url)
                        .map(this::getResponseBody)
                        .map(Optional::of)
                        .timeout(TIMEOUT_MS, TimeUnit.MILLISECONDS, Observable.empty())
                        .onErrorReturn(t -> handleException(t, result))
                )
                .doOnNext(this::logThread)
                .observeOn(Schedulers.computation())
                .doOnNext(this::logThread)
                .buffer(dbHits)
                .subscribe(v -> deferredResult.setResult(getTotalResult(v)));

        deferredResult.onCompletion(subscription::unsubscribe);

        return deferredResult;
    }

    private void logThread(Optional<String> r) {
        LOG.debug("Thread:[" + Thread.currentThread().getName()+"] : " + r);
    }

    private String getResponseBody(Response response) {
        try {
            return response.getResponseBody();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private String getTotalResult(List<Optional<String>> resultArr) {
        String totalResult = "";
        for (String r : resultArr.stream().filter(Optional::isPresent).map(Optional::get).collect(Collectors.toList()))
            totalResult += r + '\n';
        return totalResult;
    }

    private Optional<String> handleException(Throwable throwable, Result result) {
        String msg;
        if (throwable instanceof TimeoutException) {
           msg = "Request #" + throwable + " failed due to service provider not responding within the configured timeout. Url: " + result.url;
           LOG.error(msg);
        } else {
           msg = "Request #" + result.id + " failed due to error: " + throwable;
           LOG.error(msg, throwable);
        }
        return Optional.ofNullable(msg);
     }
}
{'fraction_non_alphanumeric': 0.10661401776900296, 'fraction_numerical': 0.004168037731709992, 'mean_word_length': 2.730769230769231, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 29, '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': '6005880', 'n_tokens_mistral': 2579, 'n_tokens_neox': 2497, 'n_words': 516}