<commit_before>package com.gofish.sentiment.newsanalyser;

import io.vertx.core.Future;
import io.vertx.core.http.HttpClientOptions;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.rx.java.ObservableFuture;
import io.vertx.rx.java.RxHelper;
import io.vertx.rxjava.core.AbstractVerticle;
import io.vertx.rxjava.core.buffer.Buffer;
import io.vertx.rxjava.core.eventbus.MessageConsumer;
import io.vertx.rxjava.core.http.HttpClient;
import io.vertx.rxjava.core.http.HttpClientRequest;
import rx.Single;

import java.util.Optional;
import java.util.UUID;

/**
 * @author <NAME>
 */
public class NewsAnalyserWorker extends AbstractVerticle {

    public static final String ADDRESS = "sentiment.analyser.worker";
    private static final Logger LOG = LoggerFactory.getLogger(NewsAnalyserWorker.class);

    private HttpClient httpClient;
    private MessageConsumer<JsonObject> messageConsumer;

    private String apiKey;
    private String baseUrl;
    private String urlPath;
    private Integer port;

    public NewsAnalyserWorker() {
        // Vertx requires a default constructor
    }

    public NewsAnalyserWorker(HttpClient httpClient) {
        this.httpClient = httpClient;
    }

    @Override
    public void start() throws Exception {
        LOG.info("Bringing up News Analyser Worker");

        JsonObject apiConfig = Optional.ofNullable(config().getJsonObject("api"))
                .orElseThrow(() -> new RuntimeException("Could not load analyser configuration"));

        apiKey = apiConfig.getString("key", "");
        baseUrl = apiConfig.getString("base.url", "");
        urlPath = apiConfig.getString("url.path", "");
        port = apiConfig.getInteger("port", 443);
        httpClient = Optional.ofNullable(httpClient).orElseGet(() -> vertx.createHttpClient(getHttpClientOptions()));

        messageConsumer = vertx.eventBus().localConsumer(ADDRESS, messageHandler -> {
            try {
                final JsonObject article = messageHandler.body().getJsonObject("article");
                final String analysisText = String.join(". ", article.getString("name"), article.getString("description"));
                final JsonObject requestData = new JsonObject().put("documents", new JsonArray()
                        .add(new JsonObject()
                                .put("language", "en")
                                .put("id", UUID.randomUUID().toString())
                                .put("text", analysisText)));

                final Buffer chunk = Buffer.buffer(requestData.encode());

                HttpClientRequest request = httpClient.request(HttpMethod.POST, port, baseUrl, urlPath)
                        .putHeader("Content-Type", "application/json; charset=UTF-8")
                        .putHeader("Content-Length", String.valueOf(chunk.length()))
                        .putHeader("Ocp-Apim-Subscription-Key", apiKey);

                LOG.info("Calling Text Analytics API");

                request.toObservable()
                        .flatMap(response -> {
                            ObservableFuture<JsonObject> observable = RxHelper.observableFuture();
                            response.bodyHandler(buffer -> observable.toHandler().handle(Future.succeededFuture(buffer.toJsonObject())));
                            return observable;
                        })
                        .flatMapSingle(result -> rxAddSentimentResults(article, result))
                        .subscribe(
                                result -> messageHandler.reply(result),
                                failure -> messageHandler.fail(1, failure.getMessage()),
                                () -> {
                                    // request.end() must occur inside onComplete to avoid 'Request already complete'
                                    // exceptions which may occure if initial request fails and a retry is necessary
                                    request.end();
                                    LOG.info("Finished News Analysis");
                                }
                        );

                request.write(chunk);
            }
            catch (Throwable t) {
                LOG.error(t.getMessage(), t.getCause());
                messageHandler.fail(2, "Invalid Request");
            }
        });
    }

    private Single<JsonObject> rxAddSentimentResults(JsonObject article, JsonObject analysisResponse) {
        JsonArray documents = Optional.ofNullable(analysisResponse.getJsonArray("documents"))
                .orElseThrow(() -> new RuntimeException(analysisResponse.encode()));

        article.put("sentiment", documents.getJsonObject(0));

        return Single.just(article);
    }

    /**
     * Creates and returns an HttpClientOptions object. Values for each option are retrieved from the config json object
     * (this config json object is passed in to the verticle when it is deployed)
     *
     * @return HttpClientOptions object to configure this verticles HttpClient
     */
    private HttpClientOptions getHttpClientOptions() {
        return new HttpClientOptions()
                .setPipelining(true)
                .setPipeliningLimit(10)
                .setIdleTimeout(0)
                .setSsl(true)
                .setKeepAlive(true);
    }

    @Override
    public void stop(Future<Void> stopFuture) throws Exception {
        httpClient.close();
        messageConsumer.rxUnregister().subscribe(stopFuture::complete, stopFuture::fail);
    }
}<commit_msg>Migrate NewsAnalyserWorker from HttpClient to WebClient<commit_after>package com.gofish.sentiment.newsanalyser;

import io.vertx.core.Future;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.ext.web.client.WebClientOptions;
import io.vertx.rxjava.core.AbstractVerticle;
import io.vertx.rxjava.core.eventbus.MessageConsumer;
import io.vertx.rxjava.ext.web.client.HttpRequest;
import io.vertx.rxjava.ext.web.client.HttpResponse;
import io.vertx.rxjava.ext.web.client.WebClient;
import io.vertx.rxjava.ext.web.codec.BodyCodec;
import rx.Single;

import java.util.Optional;
import java.util.UUID;

/**
 * @author <NAME>
 */
public class NewsAnalyserWorker extends AbstractVerticle {

    public static final String ADDRESS = "sentiment.analyser.worker";
    private static final Logger LOG = LoggerFactory.getLogger(NewsAnalyserWorker.class);

    private WebClient webClient;
    private MessageConsumer<JsonObject> messageConsumer;

    private String apiKey;
    private String baseUrl;
    private String urlPath;
    private Integer port;

    public NewsAnalyserWorker() {
        // Vertx requires a default constructor
    }

    public NewsAnalyserWorker(WebClient webClient) {
        this.webClient = webClient;
    }

    @Override
    public void start() throws Exception {
        LOG.info("Bringing up News Analyser Worker");

        JsonObject apiConfig = Optional.ofNullable(config().getJsonObject("api"))
                .orElseThrow(() -> new RuntimeException("Could not load analyser configuration"));

        apiKey = apiConfig.getString("key", "");
        baseUrl = apiConfig.getString("base.url", "");
        urlPath = apiConfig.getString("url.path", "");
        port = apiConfig.getInteger("port", 443);
        webClient = Optional.ofNullable(webClient).orElseGet(() -> WebClient.create(vertx, getWebClientOptions()));

        final HttpRequest<JsonObject> request = getHttpRequest();

        messageConsumer = vertx.eventBus().localConsumer(ADDRESS, messageHandler -> {
            final JsonObject article = messageHandler.body().getJsonObject("article");
            final String articleName = article.getString("name");
            final String articleDescription = article.getString("description");

            if (articleName == null && articleDescription == null) {
                messageHandler.fail(1, "Invalid Request");
            }
            else if (articleName == null) {
                messageHandler.fail(1, "Invalid article headline supplied");
            }
            else if (articleDescription == null) {
                messageHandler.fail(1, "Invalid article lead paragraph supplied");
            }

            final String analysisText = String.join(". ", articleName, articleDescription);
            final JsonObject requestData = new JsonObject().put("documents", new JsonArray()
                    .add(new JsonObject()
                            .put("language", "en")
                            .put("id", UUID.randomUUID().toString())
                            .put("text", analysisText)));

            request.rxSendJson(requestData)
                    .map(HttpResponse::body)
                    .flatMap(body -> rxAddSentimentResults(article, body))
                    .subscribe(messageHandler::reply, fail -> messageHandler.fail(1, fail.getMessage()));
        });
    }

    private Single<JsonObject> rxAddSentimentResults(JsonObject article, JsonObject analysisResponse) {
        JsonArray documents = Optional.ofNullable(analysisResponse.getJsonArray("documents"))
                .orElseThrow(() -> new RuntimeException(analysisResponse.encode()));

        article.put("sentiment", documents.getJsonObject(0));

        return Single.just(article);
    }

    private HttpRequest<JsonObject> getHttpRequest() {
        return webClient.post(port, baseUrl, urlPath)
                .putHeader("Ocp-Apim-Subscription-Key", apiKey)
                .as(BodyCodec.jsonObject());
    }

    /**
     * Creates and returns an HttpClientOptions object. Values for each option are retrieved from the config json object
     * (this config json object is passed in to the verticle when it is deployed)
     *
     * @return HttpClientOptions object to configure this verticles HttpClient
     */
    private WebClientOptions getWebClientOptions() {
        return new WebClientOptions()
                .setPipelining(true)
                .setPipeliningLimit(10)
                .setIdleTimeout(0)
                .setSsl(true)
                .setKeepAlive(true);
    }

    @Override
    public void stop(Future<Void> stopFuture) throws Exception {
        webClient.close();
        messageConsumer.rxUnregister().subscribe(stopFuture::complete, stopFuture::fail);
    }
}
{'fraction_non_alphanumeric': 0.10459497868308858, 'fraction_numerical': 0.001989578398863098, 'mean_word_length': 2.409560723514212, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7154695', 'n_tokens_mistral': 2711, 'n_tokens_neox': 2616, 'n_words': 580}