<commit_before>package com.gradle.exportapi;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
import io.reactivex.netty.protocol.http.client.HttpClient;
import io.reactivex.netty.protocol.http.client.HttpClientRequest;
import io.reactivex.netty.protocol.http.client.HttpClientResponse;
import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
import org.knowm.yank.PropertiesUtils;
import org.knowm.yank.Yank;
import rx.Observable;
import rx.exceptions.Exceptions;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.time.Duration;
import java.time.Instant;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicReference;

import static java.time.Instant.now;

/* @Author <NAME> <EMAIL> */

final class Application {

    private static final String BASIC_AUTH = System.getProperty("basic_auth");

    private static final SocketAddress GRADLE_ENTERPRISE_SERVER = new InetSocketAddress(
            System.getProperty("server"), Integer.parseInt( System.getProperty("port","443")) );

    private static final HttpClient<ByteBuf, ByteBuf> HTTP_CLIENT = HttpClient.newClient(GRADLE_ENTERPRISE_SERVER).unsafeSecure();
    private static final int THROTTLE = 5;
    private static final ObjectMapper MAPPER = new ObjectMapper();


    public static void main(String[] args) throws Exception {

        Properties dbProps = PropertiesUtils.getPropertiesFromClasspath("POSTGRES.properties");
        Yank.setupDefaultConnectionPool(dbProps);

        if( System.getProperty("createDb") != null) {
            CreateDB.run();
        }

        String hoursStr = System.getProperty("hours");

        Instant since;
        // default to 24hs
        if(hoursStr == null) {
            since = now().minus(Duration.ofHours( Integer.parseInt("24")));
        }
        else if(hoursStr.equals("all")) {
            since = Instant.EPOCH;
            System.out.println("Calculating for all stored build scans");
        } else {
            since = now().minus(Duration.ofHours( Integer.parseInt(hoursStr)));
        }



        buildIdStream(since)
                .flatMap(buildId -> buildEventStream(buildId)
                                .reduce(new EventProcessor(buildId), (eventProcessor, json) -> {
                                    eventProcessor.process(json);
                                    return eventProcessor;
                                }),
                        THROTTLE
                )

                .toBlocking()
                .subscribe(
                        System.out::println
                );

        Yank.releaseDefaultConnectionPool();
    }

    private static Observable<String> buildIdStream(Instant since) {

        return buildStream(since, null)
                .map(Application::parse)
                .map(json -> json.get("buildId").asText());
    }

    private static Observable<ServerSentEvent> buildStream(Instant since, String lastBuildEventId) {
        System.out.println("Build stream from " + lastBuildEventId);
        AtomicReference<String> lastBuildId = new AtomicReference<>(null);

        HttpClientRequest<ByteBuf, ByteBuf> request = HTTP_CLIENT
                .createGet("/build-export/v1/builds/since/" + String.valueOf(since.toEpochMilli()))
                .setKeepAlive(true);
        if(BASIC_AUTH != null) {
            request = request.addHeader("Authorization", "Basic " + BASIC_AUTH);
        }

        if (lastBuildEventId != null) {
            request.addHeader("Last-Event-ID", lastBuildEventId);
        }

        return request
                .flatMap(HttpClientResponse::getContentAsServerSentEvents)
                .doOnNext(serverSentEvent -> lastBuildId.set(serverSentEvent.getEventIdAsString()))
                .doOnSubscribe(() -> System.out.println("Streaming builds..."))
                .onErrorResumeNext(t -> {
                    System.out.println("Error streaming builds, resuming from " + lastBuildId.get() + "...");
                    return buildStream(since, lastBuildId.get());
                });
    }

    private static Observable<JsonNode> buildEventStream(String buildId) {
        return buildEventStream(buildId, null);
    }

    private static Observable<JsonNode> buildEventStream(String buildId, String lastEventId) {
        AtomicReference<String> lastBuildEventId = new AtomicReference<>(null);

        HttpClientRequest<ByteBuf, ByteBuf> request = HTTP_CLIENT
                .createGet("/build-export/v1/build/" + buildId + "/events")
                .setKeepAlive(true);
        if(BASIC_AUTH != null) {
            request = request.addHeader("Authorization", "Basic " + BASIC_AUTH);
        }

        if (lastEventId != null) {
            request.addHeader("Last-Event-ID", lastEventId);
        }

        return request
                .flatMap(HttpClientResponse::getContentAsServerSentEvents)
                .doOnNext(serverSentEvent -> lastBuildEventId.set(serverSentEvent.getEventIdAsString()))
                .doOnSubscribe(() -> System.out.println("Streaming events for : " + buildId))
                .filter(serverSentEvent -> serverSentEvent.getEventTypeAsString().equals("BuildEvent"))
                .map(Application::parse)
                .onErrorResumeNext(t -> {
                    System.out.println("Error streaming build events, resuming from " + lastBuildEventId.get() + "...");
                    return buildEventStream(buildId, lastBuildEventId.get());
                });
    }

    private static JsonNode parse(ServerSentEvent serverSentEvent) {
        try {
            return MAPPER.readTree(serverSentEvent.contentAsString());
        } catch (IOException e) {
            throw Exceptions.propagate(e);
        } finally {
            boolean deallocated = serverSentEvent.release();
            assert deallocated;
        }
    }

}
<commit_msg>Now, really, add the header for resume<commit_after>package com.gradle.exportapi;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
import io.reactivex.netty.protocol.http.client.HttpClient;
import io.reactivex.netty.protocol.http.client.HttpClientRequest;
import io.reactivex.netty.protocol.http.client.HttpClientResponse;
import io.reactivex.netty.protocol.http.sse.ServerSentEvent;
import org.knowm.yank.PropertiesUtils;
import org.knowm.yank.Yank;
import rx.Observable;
import rx.exceptions.Exceptions;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.time.Duration;
import java.time.Instant;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicReference;

import static java.time.Instant.now;

/* @Author <NAME> <EMAIL> */

final class Application {

    private static final String BASIC_AUTH = System.getProperty("basic_auth");

    private static final SocketAddress GRADLE_ENTERPRISE_SERVER = new InetSocketAddress(
            System.getProperty("server"), Integer.parseInt( System.getProperty("port","443")) );

    private static final HttpClient<ByteBuf, ByteBuf> HTTP_CLIENT = HttpClient.newClient(GRADLE_ENTERPRISE_SERVER).unsafeSecure();
    private static final int THROTTLE = 5;
    private static final ObjectMapper MAPPER = new ObjectMapper();


    public static void main(String[] args) throws Exception {

        Properties dbProps = PropertiesUtils.getPropertiesFromClasspath("POSTGRES.properties");
        Yank.setupDefaultConnectionPool(dbProps);

        if( System.getProperty("createDb") != null) {
            CreateDB.run();
        }

        String hoursStr = System.getProperty("hours");

        Instant since;
        // default to 24hs
        if(hoursStr == null) {
            since = now().minus(Duration.ofHours( Integer.parseInt("24")));
        }
        else if(hoursStr.equals("all")) {
            since = Instant.EPOCH;
            System.out.println("Calculating for all stored build scans");
        } else {
            since = now().minus(Duration.ofHours( Integer.parseInt(hoursStr)));
        }



        buildIdStream(since)
                .flatMap(buildId -> buildEventStream(buildId)
                                .reduce(new EventProcessor(buildId), (eventProcessor, json) -> {
                                    eventProcessor.process(json);
                                    return eventProcessor;
                                }),
                        THROTTLE
                )

                .toBlocking()
                .subscribe(
                        System.out::println
                );

        Yank.releaseDefaultConnectionPool();
    }

    private static Observable<String> buildIdStream(Instant since) {

        return buildStream(since, null)
                .map(Application::parse)
                .map(json -> json.get("buildId").asText());
    }

    private static Observable<ServerSentEvent> buildStream(Instant since, String lastBuildEventId) {
        System.out.println("Build stream from " + lastBuildEventId);
        AtomicReference<String> lastBuildId = new AtomicReference<>(null);

        HttpClientRequest<ByteBuf, ByteBuf> request = HTTP_CLIENT
                .createGet("/build-export/v1/builds/since/" + String.valueOf(since.toEpochMilli()))
                .setKeepAlive(true);
        if(BASIC_AUTH != null) {
            request = request.addHeader("Authorization", "Basic " + BASIC_AUTH);
        }

        if (lastBuildEventId != null) {
            request = request.addHeader("Last-Event-ID", lastBuildEventId);
        }

        return request
                .flatMap(HttpClientResponse::getContentAsServerSentEvents)
                .doOnNext(serverSentEvent -> lastBuildId.set(serverSentEvent.getEventIdAsString()))
                .doOnSubscribe(() -> System.out.println("Streaming builds..."))
                .onErrorResumeNext(t -> {
                    System.out.println("Error streaming builds, resuming from " + lastBuildId.get() + "...");
                    return buildStream(since, lastBuildId.get());
                });
    }

    private static Observable<JsonNode> buildEventStream(String buildId) {
        return buildEventStream(buildId, null);
    }

    private static Observable<JsonNode> buildEventStream(String buildId, String lastEventId) {
        AtomicReference<String> lastBuildEventId = new AtomicReference<>(null);

        HttpClientRequest<ByteBuf, ByteBuf> request = HTTP_CLIENT
                .createGet("/build-export/v1/build/" + buildId + "/events")
                .setKeepAlive(true);
        if(BASIC_AUTH != null) {
            request = request.addHeader("Authorization", "Basic " + BASIC_AUTH);
        }

        if (lastEventId != null) {
            request = request.addHeader("Last-Event-ID", lastEventId);
        }

        return request
                .flatMap(HttpClientResponse::getContentAsServerSentEvents)
                .doOnNext(serverSentEvent -> lastBuildEventId.set(serverSentEvent.getEventIdAsString()))
                .doOnSubscribe(() -> System.out.println("Streaming events for : " + buildId))
                .filter(serverSentEvent -> serverSentEvent.getEventTypeAsString().equals("BuildEvent"))
                .map(Application::parse)
                .onErrorResumeNext(t -> {
                    System.out.println("Error streaming build events, resuming from " + lastBuildEventId.get() + "...");
                    return buildEventStream(buildId, lastBuildEventId.get());
                });
    }

    private static JsonNode parse(ServerSentEvent serverSentEvent) {
        try {
            return MAPPER.readTree(serverSentEvent.contentAsString());
        } catch (IOException e) {
            throw Exceptions.propagate(e);
        } finally {
            boolean deallocated = serverSentEvent.release();
            assert deallocated;
        }
    }

}

{'fraction_non_alphanumeric': 0.10838258553233997, 'fraction_numerical': 0.0016648630650129028, 'mean_word_length': 2.4315909740074266, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3828455', 'n_tokens_mistral': 3279, 'n_tokens_neox': 3116, 'n_words': 574}