diff --git a/dashboard/server/Dockerfile b/dashboard/server/Dockerfile index c5060889b9..e0179b3a84 100644 --- a/dashboard/server/Dockerfile +++ b/dashboard/server/Dockerfile @@ -1,6 +1,6 @@ -FROM azul/zulu-openjdk-alpine:11 +FROM azul/zulu-openjdk-alpine:19 ARG DEPENDENCY=dependency COPY ${DEPENDENCY}/BOOT-INF/lib /app/lib COPY ${DEPENDENCY}/META-INF /app/META-INF COPY ${DEPENDENCY}/BOOT-INF/classes /app -ENTRYPOINT ["java", "-cp", "/app:/app/lib/*", "build.bazel.dashboard.Application"] +ENTRYPOINT ["java", "--enable-preview", "--add-modules=jdk.incubator.concurrent", "-Djdk.tracePinnedThreads", "-cp", "/app:/app/lib/*", "build.bazel.dashboard.Application"] diff --git a/dashboard/server/pom.xml b/dashboard/server/pom.xml index b60a3961cd..0ce621f627 100644 --- a/dashboard/server/pom.xml +++ b/dashboard/server/pom.xml @@ -5,7 +5,7 @@ org.springframework.boot spring-boot-starter-parent - 2.6.6 + 3.0.0 build.bazel @@ -15,7 +15,7 @@ The server for the dashboard - 11 + 19 DEV @@ -46,13 +46,12 @@ caffeine - io.r2dbc + org.postgresql r2dbc-postgresql io.reactivex.rxjava3 rxjava - 3.0.13 io.projectreactor.addons @@ -62,6 +61,11 @@ org.springframework.boot spring-boot-starter-mail + + org.springframework.boot + spring-boot-properties-migrator + runtime + org.springframework.boot @@ -98,9 +102,29 @@ + + org.apache.maven.plugins + maven-compiler-plugin + + + --enable-preview + --add-modules=jdk.incubator.concurrent + + + + + org.apache.maven.plugins + maven-surefire-plugin + + --enable-preview --add-modules=jdk.incubator.concurrent + + org.springframework.boot spring-boot-maven-plugin + + --enable-preview --add-modules=jdk.incubator.concurrent -Djdk.tracePinnedThreads + diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/api/GithubApi.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/api/GithubApi.java index 5d9efe4fd0..630c9ed046 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/api/GithubApi.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/api/GithubApi.java @@ -1,17 +1,15 @@ package build.bazel.dashboard.github.api; -import io.reactivex.rxjava3.core.Single; - public interface GithubApi { - Single listRepositoryIssues(ListRepositoryIssuesRequest request); + GithubApiResponse listRepositoryIssues(ListRepositoryIssuesRequest request); - Single listRepositoryEvents(ListRepositoryEventsRequest request); + GithubApiResponse listRepositoryEvents(ListRepositoryEventsRequest request); - Single listRepositoryIssueEvents(ListRepositoryIssueEventsRequest request); + GithubApiResponse listRepositoryIssueEvents(ListRepositoryIssueEventsRequest request); - Single fetchIssue(FetchIssueRequest request); + GithubApiResponse fetchIssue(FetchIssueRequest request); - Single listIssueComments(ListIssueCommentsRequest request); + GithubApiResponse listIssueComments(ListIssueCommentsRequest request); - Single searchIssues(SearchIssuesRequest request); + GithubApiResponse searchIssues(SearchIssuesRequest request); } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/api/GithubApiResponse.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/api/GithubApiResponse.java index 3bdc5b3be7..3923f3d7e7 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/api/GithubApiResponse.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/api/GithubApiResponse.java @@ -1,19 +1,19 @@ package build.bazel.dashboard.github.api; +import static build.bazel.dashboard.utils.HttpHeadersUtils.getAsIntOrZero; +import static build.bazel.dashboard.utils.HttpHeadersUtils.getOrEmpty; + import com.fasterxml.jackson.databind.JsonNode; import lombok.Builder; import lombok.Value; -import org.springframework.http.HttpStatus; +import org.springframework.http.HttpStatusCode; import org.springframework.web.reactive.function.client.ClientResponse; import reactor.core.publisher.Mono; -import static build.bazel.dashboard.utils.HttpHeadersUtils.getAsIntOrZero; -import static build.bazel.dashboard.utils.HttpHeadersUtils.getOrEmpty; - @Builder @Value public class GithubApiResponse { - HttpStatus status; + HttpStatusCode status; String etag; RateLimit rateLimit; JsonNode body; @@ -37,7 +37,7 @@ public static RateLimit fromHeaders(ClientResponse.Headers headers) { } public static Mono fromClientResponse(ClientResponse clientResponse) { - HttpStatus status = clientResponse.statusCode(); + var status = clientResponse.statusCode(); ClientResponse.Headers headers = clientResponse.headers(); GithubApiResponseBuilder builder = GithubApiResponse.builder() diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/api/WebClientGithubApi.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/api/WebClientGithubApi.java index 0de1b4b904..196b8d7bc0 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/api/WebClientGithubApi.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/api/WebClientGithubApi.java @@ -1,20 +1,16 @@ package build.bazel.dashboard.github.api; +import static com.google.common.base.Preconditions.checkNotNull; + import com.google.common.base.Strings; -import io.reactivex.rxjava3.core.Single; +import java.net.URI; +import java.util.Optional; +import java.util.function.Function; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.util.UriBuilder; -import reactor.adapter.rxjava.RxJava3Adapter; - -import java.net.URI; -import java.util.Optional; -import java.util.function.Function; - -import static com.google.common.base.Preconditions.checkNotNull; -import static java.nio.charset.StandardCharsets.UTF_8; @Component @Slf4j @@ -32,7 +28,7 @@ public WebClientGithubApi( } @Override - public Single listRepositoryIssues(ListRepositoryIssuesRequest request) { + public GithubApiResponse listRepositoryIssues(ListRepositoryIssuesRequest request) { log.debug("{}", request); checkNotNull(request.getOwner()); @@ -45,13 +41,13 @@ public Single listRepositoryIssues(ListRepositoryIssuesReques .pathSegment("repos", request.getOwner(), request.getRepo(), "issues") .queryParamIfPresent("per_page", Optional.ofNullable(request.getPerPage())) .queryParamIfPresent("page", Optional.ofNullable(request.getPage())) - .build()); + .build(), ""); return exchange(spec); } @Override - public Single listRepositoryEvents(ListRepositoryEventsRequest request) { + public GithubApiResponse listRepositoryEvents(ListRepositoryEventsRequest request) { log.debug("{}", request); checkNotNull(request.getOwner()); @@ -64,16 +60,13 @@ public Single listRepositoryEvents(ListRepositoryEventsReques .pathSegment("repos", request.getOwner(), request.getRepo(), "events") .queryParamIfPresent("per_page", Optional.ofNullable(request.getPerPage())) .queryParamIfPresent("page", Optional.ofNullable(request.getPage())) - .build()); - if (!Strings.isNullOrEmpty(request.getEtag())) { - spec.ifNoneMatch(request.getEtag()); - } + .build(), request.getEtag()); return exchange(spec); } @Override - public Single listRepositoryIssueEvents(ListRepositoryIssueEventsRequest request) { + public GithubApiResponse listRepositoryIssueEvents(ListRepositoryIssueEventsRequest request) { log.debug("{}", request); checkNotNull(request.getOwner()); @@ -86,16 +79,13 @@ public Single listRepositoryIssueEvents(ListRepositoryIssueEv .pathSegment("repos", request.getOwner(), request.getRepo(), "issues", "events") .queryParamIfPresent("per_page", Optional.ofNullable(request.getPerPage())) .queryParamIfPresent("page", Optional.ofNullable(request.getPage())) - .build()); - if (!Strings.isNullOrEmpty(request.getEtag())) { - spec.ifNoneMatch(request.getEtag()); - } + .build(), request.getEtag()); return exchange(spec); } @Override - public Single fetchIssue(FetchIssueRequest request) { + public GithubApiResponse fetchIssue(FetchIssueRequest request) { log.debug("{}", request); checkNotNull(request.getOwner()); @@ -111,16 +101,13 @@ public Single fetchIssue(FetchIssueRequest request) { request.getRepo(), "issues", String.valueOf(request.getIssueNumber())) - .build()); - if (!Strings.isNullOrEmpty(request.getEtag())) { - spec.ifNoneMatch(request.getEtag()); - } + .build(), request.getEtag()); return exchange(spec); } @Override - public Single listIssueComments(ListIssueCommentsRequest request) { + public GithubApiResponse listIssueComments(ListIssueCommentsRequest request) { log.debug("{}", request); checkNotNull(request.getOwner()); @@ -130,19 +117,22 @@ public Single listIssueComments(ListIssueCommentsRequest requ get( uriBuilder -> uriBuilder - .pathSegment("repos", request.getOwner(), request.getRepo(), "issues", Integer.toString(request.getIssueNumber()), "comments") + .pathSegment( + "repos", + request.getOwner(), + request.getRepo(), + "issues", + Integer.toString(request.getIssueNumber()), + "comments") .queryParamIfPresent("per_page", Optional.ofNullable(request.getPerPage())) .queryParamIfPresent("page", Optional.ofNullable(request.getPage())) - .build()); - if (!Strings.isNullOrEmpty(request.getEtag())) { - spec.ifNoneMatch(request.getEtag()); - } + .build(), request.getEtag()); return exchange(spec); } @Override - public Single searchIssues(SearchIssuesRequest request) { + public GithubApiResponse searchIssues(SearchIssuesRequest request) { log.debug("{}", request); checkNotNull(request.getQ()); @@ -157,18 +147,22 @@ public Single searchIssues(SearchIssuesRequest request) { .queryParamIfPresent("order", Optional.ofNullable(request.getOrder())) .queryParamIfPresent("per_page", Optional.ofNullable(request.getPerPage())) .queryParamIfPresent("page", Optional.ofNullable(request.getPage())) - .build(request.getQ())); + .build(request.getQ()), ""); return exchange(spec); } - private WebClient.RequestHeadersSpec get(Function uriFunction) { + private WebClient.RequestHeadersSpec get(Function uriFunction, String etag) { WebClient.RequestHeadersSpec spec = webClient .get() .uri(uriBuilder -> uriFunction.apply(uriBuilder.scheme(SCHEME).host(HOST))) .header("Accept", "application/vnd.github.v3+json"); + if (!Strings.isNullOrEmpty(etag)) { + spec.ifNoneMatch(etag); + } + if (!Strings.isNullOrEmpty(accessToken)) { spec = spec.header("Authorization", "token " + accessToken); } @@ -176,7 +170,10 @@ private WebClient.RequestHeadersSpec get(Function uriFunctio return spec; } - private Single exchange(WebClient.RequestHeadersSpec spec) { - return RxJava3Adapter.monoToSingle(spec.exchangeToMono(GithubApiResponse::fromClientResponse)); + private GithubApiResponse exchange(WebClient.RequestHeadersSpec spec) { + return spec.exchangeToMono(response -> { + log.debug("{} {}", response.statusCode(), response.headers().asHttpHeaders().toSingleValueMap()); + return GithubApiResponse.fromClientResponse(response); + }).block(); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssue.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssue.java index a9867f01d8..390f342d98 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssue.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssue.java @@ -3,14 +3,13 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.PropertyNamingStrategy; +import com.fasterxml.jackson.databind.PropertyNamingStrategies; import com.fasterxml.jackson.databind.annotation.JsonNaming; -import lombok.Builder; -import lombok.Value; - -import javax.annotation.Nullable; import java.time.Instant; import java.util.List; +import javax.annotation.Nullable; +import lombok.Builder; +import lombok.Value; @Builder @Value @@ -42,7 +41,7 @@ public Data parseData(ObjectMapper objectMapper) throws JsonProcessingException @Builder @Value - @JsonNaming(PropertyNamingStrategy.SnakeCaseStrategy.class) + @JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class) public static class Data { int number; User user; @@ -56,7 +55,7 @@ public static class Data { @Builder @Value - @JsonNaming(PropertyNamingStrategy.SnakeCaseStrategy.class) + @JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class) public static class User { String login; String avatarUrl; @@ -64,7 +63,7 @@ public static class User { @Builder @Value - @JsonNaming(PropertyNamingStrategy.SnakeCaseStrategy.class) + @JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class) public static class Label { String name; String description; diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRepo.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRepo.java index 8cb12065f7..42041594c1 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRepo.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRepo.java @@ -1,19 +1,17 @@ package build.bazel.dashboard.github.issue; -import io.reactivex.rxjava3.core.Completable; -import io.reactivex.rxjava3.core.Maybe; import io.reactivex.rxjava3.core.Observable; -import io.reactivex.rxjava3.core.Single; +import java.util.Optional; public interface GithubIssueRepo { - Completable save(GithubIssue githubIssue); + void save(GithubIssue githubIssue); - Completable delete(String owner, String repo, int issueNumber); + void delete(String owner, String repo, int issueNumber); - Maybe findOne(String owner, String repo, int issueNumber); + Optional findOne(String owner, String repo, int issueNumber); Observable list(); - Single findMaxIssueNumber(String owner, String repo); + Integer findMaxIssueNumber(String owner, String repo); } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRepoPg.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRepoPg.java index aeca438557..78269322ec 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRepoPg.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRepoPg.java @@ -1,24 +1,20 @@ package build.bazel.dashboard.github.issue; -import com.fasterxml.jackson.core.JsonProcessingException; +import static build.bazel.dashboard.utils.PgJson.toPgJson; +import static java.util.Objects.requireNonNull; + import com.fasterxml.jackson.databind.ObjectMapper; import io.r2dbc.postgresql.codec.Json; -import io.r2dbc.spi.Row; -import io.reactivex.rxjava3.core.Completable; -import io.reactivex.rxjava3.core.Maybe; +import io.r2dbc.spi.Readable; import io.reactivex.rxjava3.core.Observable; -import io.reactivex.rxjava3.core.Single; +import java.io.IOException; +import java.time.Instant; +import java.util.Optional; import lombok.RequiredArgsConstructor; import org.springframework.r2dbc.core.DatabaseClient; import org.springframework.stereotype.Repository; import reactor.adapter.rxjava.RxJava3Adapter; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import java.io.IOException; -import java.time.Instant; - -import static java.util.Objects.requireNonNull; @Repository @RequiredArgsConstructor @@ -28,45 +24,39 @@ public class GithubIssueRepoPg implements GithubIssueRepo { private final ObjectMapper objectMapper; @Override - public Completable save(GithubIssue githubIssue) { - try { - Mono execution = - databaseClient - .sql( - "INSERT INTO github_issue_data (owner, repo, issue_number, timestamp, etag, data)" - + " VALUES (:owner, :repo, :issue_number, :timestamp, :etag, :data) ON" - + " CONFLICT (owner, repo, issue_number) DO UPDATE SET etag = excluded.etag," - + " timestamp = excluded.timestamp, data = excluded.data") - .bind("owner", githubIssue.getOwner()) - .bind("repo", githubIssue.getRepo()) - .bind("issue_number", githubIssue.getIssueNumber()) - .bind("timestamp", githubIssue.getTimestamp()) - .bind("etag", githubIssue.getEtag()) - .bind("data", Json.of(objectMapper.writeValueAsBytes(githubIssue.getData()))) - .then(); - return RxJava3Adapter.monoToCompletable(execution); - } catch (JsonProcessingException e) { - return Completable.error(e); - } + public void save(GithubIssue githubIssue) { + databaseClient + .sql( + "INSERT INTO github_issue_data (owner, repo, issue_number, timestamp, etag, data)" + + " VALUES (:owner, :repo, :issue_number, :timestamp, :etag, :data) ON" + + " CONFLICT (owner, repo, issue_number) DO UPDATE SET etag = excluded.etag," + + " timestamp = excluded.timestamp, data = excluded.data") + .bind("owner", githubIssue.getOwner()) + .bind("repo", githubIssue.getRepo()) + .bind("issue_number", githubIssue.getIssueNumber()) + .bind("timestamp", githubIssue.getTimestamp()) + .bind("etag", githubIssue.getEtag()) + .bind("data", toPgJson(objectMapper, githubIssue.getData())) + .then() + .block(); } @Override - public Completable delete(String owner, String repo, int issueNumber) { - Mono execution = - databaseClient - .sql( - "DELETE FROM github_issue_data WHERE owner = :owner AND repo = :repo AND" - + " issue_number = :issue_number") - .bind("owner", owner) - .bind("repo", repo) - .bind("issue_number", issueNumber) - .then(); - return RxJava3Adapter.monoToCompletable(execution); + public void delete(String owner, String repo, int issueNumber) { + databaseClient + .sql( + "DELETE FROM github_issue_data WHERE owner = :owner AND repo = :repo AND" + + " issue_number = :issue_number") + .bind("owner", owner) + .bind("repo", repo) + .bind("issue_number", issueNumber) + .then() + .block(); } @Override - public Maybe findOne(String owner, String repo, int issueNumber) { - Mono query = + public Optional findOne(String owner, String repo, int issueNumber) { + return Optional.ofNullable( databaseClient .sql( "SELECT owner, repo, issue_number, timestamp, etag, data FROM github_issue_data " @@ -75,8 +65,8 @@ public Maybe findOne(String owner, String repo, int issueNumber) { .bind("repo", repo) .bind("issue_number", issueNumber) .map(this::toGithubIssue) - .one(); - return RxJava3Adapter.monoToMaybe(query); + .one() + .block()); } @Override @@ -90,7 +80,7 @@ public Observable list() { return RxJava3Adapter.fluxToObservable(query); } - private GithubIssue toGithubIssue(Row row) { + private GithubIssue toGithubIssue(Readable row) { try { return GithubIssue.builder() .owner(requireNonNull(row.get("owner", String.class))) @@ -106,16 +96,15 @@ private GithubIssue toGithubIssue(Row row) { } @Override - public Single findMaxIssueNumber(String owner, String repo) { - Mono query = - databaseClient - .sql( - "SELECT COALESCE(MAX(issue_number), 0) as max_issue_number FROM github_issue WHERE owner =" - + " :owner AND repo = :repo") - .bind("owner", owner) - .bind("repo", repo) - .map(row -> row.get("max_issue_number", Integer.class)) - .one(); - return RxJava3Adapter.monoToSingle(query); + public Integer findMaxIssueNumber(String owner, String repo) { + return databaseClient + .sql( + "SELECT COALESCE(MAX(issue_number), 0) as max_issue_number FROM github_issue WHERE" + + " owner = :owner AND repo = :repo") + .bind("owner", owner) + .bind("repo", repo) + .map(row -> row.get("max_issue_number", Integer.class)) + .one() + .block(); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRestController.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRestController.java index 99f7a931ae..805565ca2b 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRestController.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueRestController.java @@ -1,8 +1,12 @@ package build.bazel.dashboard.github.issue; +import static build.bazel.dashboard.utils.RxJavaVirtualThread.maybe; +import static build.bazel.dashboard.utils.RxJavaVirtualThread.single; + import build.bazel.dashboard.github.issuecomment.GithubIssueCommentService; import io.reactivex.rxjava3.core.Maybe; import io.reactivex.rxjava3.core.Single; +import java.util.Optional; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.web.bind.annotation.GetMapping; @@ -22,10 +26,10 @@ public Maybe findOneGithubIssue( @PathVariable("owner") String owner, @PathVariable("repo") String repo, @PathVariable("issueNumber") Integer issueNumber) { - return githubIssueService - .fetchAndSave(owner, repo, issueNumber) - .map(GithubIssueService.FetchResult::getIssue) - .toMaybe(); + return maybe( + () -> + Optional.ofNullable( + githubIssueService.fetchAndSave(owner, repo, issueNumber).getIssue())); } @PutMapping("/internal/github/{owner}/{repo}/issues/{issueNumber}") @@ -33,13 +37,12 @@ public Single updateGithubIssue( @PathVariable("owner") String owner, @PathVariable("repo") String repo, @PathVariable("issueNumber") Integer issueNumber) { - return githubIssueService - .fetchAndSave(owner, repo, issueNumber) - .flatMap( - result -> - githubIssueCommentService - .syncIssueComments(owner, repo, issueNumber) - .toSingleDefault(result)); + return single( + () -> { + var result = githubIssueService.fetchAndSave(owner, repo, issueNumber); + githubIssueCommentService.syncIssueComments(owner, repo, issueNumber); + return result; + }); } /* diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueService.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueService.java index 3a84b59571..06e9f35fca 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueService.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issue/GithubIssueService.java @@ -4,17 +4,15 @@ import build.bazel.dashboard.github.api.GithubApi; import build.bazel.dashboard.github.issuestatus.GithubIssueStatus; import build.bazel.dashboard.github.issuestatus.GithubIssueStatusService; -import io.reactivex.rxjava3.core.Maybe; -import io.reactivex.rxjava3.core.Single; +import java.io.IOException; +import java.time.Instant; +import java.util.Optional; import lombok.Builder; import lombok.RequiredArgsConstructor; import lombok.Value; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; -import java.io.IOException; -import java.time.Instant; - @Service @Slf4j @RequiredArgsConstructor @@ -50,95 +48,71 @@ static FetchResult create( } } - public Maybe findOne(String owner, String repo, int issueNumber) { + public Optional findOne(String owner, String repo, int issueNumber) { return githubIssueRepo.findOne(owner, repo, issueNumber); } - public Single fetchAndSave(String owner, String repo, int issueNumber) { - return githubIssueRepo - .findOne(owner, repo, issueNumber) - .switchIfEmpty(Single.just(GithubIssue.empty(owner, repo, issueNumber))) - .flatMap( - existed -> { - FetchIssueRequest request = - FetchIssueRequest.builder() - .owner(owner) - .repo(repo) - .issueNumber(issueNumber) - .etag(existed.getEtag()) - .build(); - boolean exists = existed.getTimestamp().isAfter(Instant.EPOCH); + public FetchResult fetchAndSave(String owner, String repo, int issueNumber) { + var existed = + githubIssueRepo + .findOne(owner, repo, issueNumber) + .orElse(GithubIssue.empty(owner, repo, issueNumber)); - return githubApi - .fetchIssue(request) - .flatMap( - response -> { - if (response.getStatus().is2xxSuccessful()) { - GithubIssue githubIssue = - GithubIssue.builder() - .owner(owner) - .repo(repo) - .issueNumber(issueNumber) - .timestamp(Instant.now()) - .etag(response.getEtag()) - .data(response.getBody()) - .build(); - return githubIssueRepo - .save(githubIssue) - .andThen(githubIssueStatusService.check(githubIssue, Instant.now())) - .map( - status -> - FetchResult.create( - githubIssue, status, !exists, exists, false, null)) - .switchIfEmpty( - Single.just( - FetchResult.create( - githubIssue, null, !exists, exists, false, null))); - } else if (response.getStatus().value() == 304) { - // Not modified - return githubIssueStatusService - .check(existed, Instant.now()) - .map( - status -> - FetchResult.create( - existed, status, false, false, false, null)) - .switchIfEmpty( - Single.just( - FetchResult.create( - existed, null, false, false, false, null))); - } else if (response.getStatus().value() == 301 - || response.getStatus().value() == 404 - || response.getStatus().value() == 410) { - // Transferred or deleted - return githubIssueRepo - .delete(owner, repo, issueNumber) - // Mark existing status to DELETED - .andThen( - githubIssueStatusService.markDeleted(owner, repo, issueNumber)) - .toSingle( - () -> - FetchResult.create(existed, null, false, false, true, null)); - } else { - log.error( - "Failed to fetch {}/{}/issues/{}: {}", - owner, - repo, - issueNumber, - response.getStatus().toString()); - return Single.just( - FetchResult.create( - existed, - null, - false, - false, - false, - new IOException(response.getStatus().toString()))); - } - }); - }); + FetchIssueRequest request = + FetchIssueRequest.builder() + .owner(owner) + .repo(repo) + .issueNumber(issueNumber) + .etag(existed.getEtag()) + .build(); + boolean exists = existed.getTimestamp().isAfter(Instant.EPOCH); + var response = githubApi.fetchIssue(request); + if (response.getStatus().is2xxSuccessful()) { + GithubIssue githubIssue = + GithubIssue.builder() + .owner(owner) + .repo(repo) + .issueNumber(issueNumber) + .timestamp(Instant.now()) + .etag(response.getEtag()) + .data(response.getBody()) + .build(); + try { + githubIssueRepo.save(githubIssue); + var status = githubIssueStatusService.check(githubIssue, Instant.now()); + return FetchResult.create(githubIssue, status.orElse(null), !exists, exists, false, null); + } catch (IOException e) { + return FetchResult.create(githubIssue, null, !exists, exists, false, e); + } + } else if (response.getStatus().value() == 304) { + // Not modified + try { + var status = githubIssueStatusService.check(existed, Instant.now()); + return FetchResult.create(existed, status.orElse(null), false, false, false, null); + } catch (IOException e) { + return FetchResult.create(existed, null, false, false, false, e); + } + } else if (response.getStatus().value() == 301 + || response.getStatus().value() == 404 + || response.getStatus().value() == 410) { + // Transferred or deleted + githubIssueRepo.delete(owner, repo, issueNumber); + // Mark existing status to DELETED + githubIssueStatusService.markDeleted(owner, repo, issueNumber); + return FetchResult.create(existed, null, false, false, true, null); + } else { + log.error( + "Failed to fetch {}/{}/issues/{}: {}", + owner, + repo, + issueNumber, + response.getStatus().toString()); + return FetchResult.create( + existed, null, false, false, false, new IOException(response.getStatus().toString())); + } } - public Single findMaxIssueNumber(String owner, String repo) { + public Integer findMaxIssueNumber(String owner, String repo) { return githubIssueRepo.findMaxIssueNumber(owner, repo); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubComment.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubComment.java index e5f8863fb8..8fabae45e5 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubComment.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubComment.java @@ -1,14 +1,14 @@ package build.bazel.dashboard.github.issuecomment; import build.bazel.dashboard.github.issue.GithubIssue; -import com.fasterxml.jackson.databind.PropertyNamingStrategy; +import com.fasterxml.jackson.databind.PropertyNamingStrategies; import com.fasterxml.jackson.databind.annotation.JsonNaming; import lombok.Builder; import lombok.Value; @Builder @Value -@JsonNaming(PropertyNamingStrategy.SnakeCaseStrategy.class) +@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class) public class GithubComment { long id; String body; diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRepo.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRepo.java index 3df94da377..a8926dc748 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRepo.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRepo.java @@ -1,10 +1,9 @@ package build.bazel.dashboard.github.issuecomment; import com.fasterxml.jackson.databind.JsonNode; -import io.reactivex.rxjava3.core.Completable; import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.core.Maybe; import java.time.Instant; +import java.util.Optional; import lombok.Builder; import lombok.Data; @@ -21,9 +20,9 @@ class GithubIssueCommentPage { JsonNode data; } - Maybe findOnePage(String owner, String repo, int issueNumber, int page); + Optional findOnePage(String owner, String repo, int issueNumber, int page); Flowable findAllPages(String owner, String repo, int issueNumber); - Completable savePage(GithubIssueCommentPage page); + void savePage(GithubIssueCommentPage page); } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRepoPg.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRepoPg.java index 08959d0490..bfdf1cc8a2 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRepoPg.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRepoPg.java @@ -1,16 +1,15 @@ package build.bazel.dashboard.github.issuecomment; +import static build.bazel.dashboard.utils.PgJson.toPgJson; import static java.util.Objects.requireNonNull; -import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import io.r2dbc.postgresql.codec.Json; -import io.r2dbc.spi.Row; -import io.reactivex.rxjava3.core.Completable; +import io.r2dbc.spi.Readable; import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.core.Maybe; import java.io.IOException; import java.time.Instant; +import java.util.Optional; import lombok.RequiredArgsConstructor; import org.springframework.r2dbc.core.DatabaseClient; import org.springframework.stereotype.Repository; @@ -26,7 +25,7 @@ public class GithubIssueCommentRepoPg implements GithubIssueCommentRepo { private final ObjectMapper objectMapper; @Override - public Maybe findOnePage( + public Optional findOnePage( String owner, String repo, int issueNumber, int page) { Mono query = databaseClient @@ -40,7 +39,7 @@ public Maybe findOnePage( .bind("page", page) .map(this::toGithubIssueCommentPage) .one(); - return RxJava3Adapter.monoToMaybe(query); + return Optional.ofNullable(query.block()); } @Override @@ -59,7 +58,7 @@ public Flowable findAllPages(String owner, String repo, return RxJava3Adapter.fluxToFlowable(query); } - private GithubIssueCommentPage toGithubIssueCommentPage(Row row) { + private GithubIssueCommentPage toGithubIssueCommentPage(Readable row) { try { return GithubIssueCommentPage.builder() .owner(requireNonNull(row.get("owner", String.class))) @@ -76,27 +75,23 @@ private GithubIssueCommentPage toGithubIssueCommentPage(Row row) { } @Override - public Completable savePage(GithubIssueCommentPage page) { - try { - Mono execution = - databaseClient - .sql( - "INSERT INTO github_issue_comment_data (owner, repo, issue_number, page," - + " timestamp, etag, data) VALUES (:owner, :repo, :issue_number, :page," - + " :timestamp, :etag, :data) ON CONFLICT (owner, repo, issue_number, page)" - + " DO UPDATE SET etag = excluded.etag, timestamp = excluded.timestamp, data" - + " = excluded.data") - .bind("owner", page.getOwner()) - .bind("repo", page.getRepo()) - .bind("issue_number", page.getIssueNumber()) - .bind("page", page.getPage()) - .bind("timestamp", page.getTimestamp()) - .bind("etag", page.getEtag()) - .bind("data", Json.of(objectMapper.writeValueAsBytes(page.getData()))) - .then(); - return RxJava3Adapter.monoToCompletable(execution); - } catch (JsonProcessingException e) { - return Completable.error(e); - } + public void savePage(GithubIssueCommentPage page) { + Mono execution = + databaseClient + .sql( + "INSERT INTO github_issue_comment_data (owner, repo, issue_number, page," + + " timestamp, etag, data) VALUES (:owner, :repo, :issue_number, :page," + + " :timestamp, :etag, :data) ON CONFLICT (owner, repo, issue_number, page)" + + " DO UPDATE SET etag = excluded.etag, timestamp = excluded.timestamp, data" + + " = excluded.data") + .bind("owner", page.getOwner()) + .bind("repo", page.getRepo()) + .bind("issue_number", page.getIssueNumber()) + .bind("page", page.getPage()) + .bind("timestamp", page.getTimestamp()) + .bind("etag", page.getEtag()) + .bind("data", toPgJson(objectMapper, page.getData())) + .then(); + execution.block(); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRestController.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRestController.java index 1e9051e8b3..35c3999aea 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRestController.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentRestController.java @@ -1,5 +1,7 @@ package build.bazel.dashboard.github.issuecomment; +import static build.bazel.dashboard.utils.RxJavaVirtualThread.completable; + import io.reactivex.rxjava3.core.Completable; import io.reactivex.rxjava3.core.Flowable; import lombok.RequiredArgsConstructor; @@ -20,7 +22,7 @@ public Completable updateGithubIssueComment( @PathVariable("owner") String owner, @PathVariable("repo") String repo, @PathVariable("issueNumber") Integer issueNumber) { - return githubIssueCommentService.syncIssueComments(owner, repo, issueNumber); + return completable(() -> githubIssueCommentService.syncIssueComments(owner, repo, issueNumber)); } @GetMapping("/internal/github/{owner}/{repo}/issues/{issueNumber}/comments") diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentService.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentService.java index 73fda5f19a..4a523890f6 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentService.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuecomment/GithubIssueCommentService.java @@ -5,13 +5,9 @@ import build.bazel.dashboard.github.issuecomment.GithubIssueCommentRepo.GithubIssueCommentPage; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import io.reactivex.rxjava3.core.Completable; import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.core.Single; import java.io.IOException; import java.time.Instant; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; @@ -20,6 +16,7 @@ @RequiredArgsConstructor @Slf4j public class GithubIssueCommentService { + private static final int PER_PAGE = 100; private final GithubIssueCommentRepo githubIssueCommentRepo; @@ -33,71 +30,58 @@ public Flowable findIssueComments(String owner, String repo, int .map(jsonNode -> objectMapper.treeToValue(jsonNode, GithubComment.class)); } - public Completable syncIssueComments(String owner, String repo, int issueNumber) { - Flowable pages = - Flowable.generate( - AtomicInteger::new, - (state, emitter) -> { - emitter.onNext(state.incrementAndGet()); - }); - - return Completable.fromPublisher( - pages - .flatMapSingle( - page -> syncIssueCommentPage(owner, repo, issueNumber, page), false, 1) - .takeUntil(node -> node.size() < PER_PAGE)) - .onErrorComplete( - error -> { - log.error("Failed to sync issue comments: " + error.getMessage(), error); - return true; - }); + public void syncIssueComments(String owner, String repo, int issueNumber) { + int page = 1; + while (true) { + try { + var node = syncIssueCommentPage(owner, repo, issueNumber, page); + if (node.size() < PER_PAGE) { + break; + } + } catch (IOException e) { + log.error("Failed to sync issue comments: " + e.getMessage(), e); + return; + } + } } - private Single syncIssueCommentPage( - String owner, String repo, int issueNumber, int page) { - AtomicReference existedPage = new AtomicReference<>(); - return githubIssueCommentRepo - .findOnePage(owner, repo, issueNumber, page) - .doOnSuccess(existedPage::set) - .map(GithubIssueCommentPage::getEtag) - .defaultIfEmpty("") - .map( - etag -> - ListIssueCommentsRequest.builder() - .owner(owner) - .repo(repo) - .issueNumber(issueNumber) - .perPage(PER_PAGE) - .page(page) - .etag(etag) - .build()) - .flatMap(githubApi::listIssueComments) - .flatMap( - response -> { - if (response.getStatus().is2xxSuccessful()) { - String etag = response.getEtag(); - return githubIssueCommentRepo - .savePage( - GithubIssueCommentPage.builder() - .owner(owner) - .repo(repo) - .issueNumber(issueNumber) - .page(page) - .timestamp(Instant.now()) - .etag(etag) - .data(response.getBody()) - .build()) - .andThen(Single.just(response.getBody())); - } else if (response.getStatus().value() == 304) { - // Not modified - GithubIssueCommentPage existed = existedPage.get(); - if (existed == null) { - return Single.just(objectMapper.createArrayNode()); - } - return Single.just(existed.getData()); - } else { - return Single.error(new IOException(response.getStatus().toString())); - } - }); + private JsonNode syncIssueCommentPage(String owner, String repo, int issueNumber, int page) + throws IOException { + var existedPage = githubIssueCommentRepo.findOnePage(owner, repo, issueNumber, page); + var existedEtag = existedPage.map(GithubIssueCommentPage::getEtag).orElse(""); + var request = + ListIssueCommentsRequest.builder() + .owner(owner) + .repo(repo) + .issueNumber(issueNumber) + .perPage(PER_PAGE) + .page(page) + .etag(existedEtag) + .build(); + var response = githubApi.listIssueComments(request); + if (response.getStatus().is2xxSuccessful()) { + githubIssueCommentRepo.savePage( + GithubIssueCommentPage.builder() + .owner(owner) + .repo(repo) + .issueNumber(issueNumber) + .page(page) + .timestamp(Instant.now()) + .etag(response.getEtag()) + .data(response.getBody()) + .build()); + return response.getBody(); + } else if (response.getStatus().value() == 304) { + // Not modified + if (existedPage.isPresent()) { + return existedPage.get().getData(); + } + return objectMapper.createArrayNode(); + } else if (response.getStatus().value() == 404) { + // Not found + return objectMapper.createArrayNode(); + } else { + throw new IOException(response.getStatus().toString()); + } } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRepo.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRepo.java index dc3489f8df..be8a5ae284 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRepo.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRepo.java @@ -1,15 +1,17 @@ package build.bazel.dashboard.github.issuelist; import build.bazel.dashboard.github.issuelist.GithubIssueListService.ListParams; +import com.google.common.collect.ImmutableList; import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.core.Single; +import java.util.List; public interface GithubIssueListRepo { Flowable find(ListParams params); Single count(ListParams params); - Flowable findAllActionOwner(ListParams params); + ImmutableList findAllActionOwner(ListParams params); - Flowable findAllLabels(ListParams params); + ImmutableList findAllLabels(ListParams params); } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRepoPg.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRepoPg.java index c6a82498ad..017134d5d0 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRepoPg.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRepoPg.java @@ -1,12 +1,20 @@ package build.bazel.dashboard.github.issuelist; +import static com.google.common.collect.ImmutableList.toImmutableList; +import static java.util.Objects.requireNonNull; + import build.bazel.dashboard.github.issuelist.GithubIssueListService.ListParams; import build.bazel.dashboard.github.issuestatus.GithubIssueStatus; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.ImmutableList; import io.r2dbc.postgresql.codec.Json; -import io.r2dbc.spi.Row; +import io.r2dbc.spi.Readable; import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.core.Single; +import java.io.IOException; +import java.time.Instant; +import java.util.HashMap; +import java.util.Map; import lombok.Builder; import lombok.RequiredArgsConstructor; import org.springframework.r2dbc.core.DatabaseClient; @@ -14,13 +22,6 @@ import org.springframework.stereotype.Repository; import reactor.adapter.rxjava.RxJava3Adapter; -import java.io.IOException; -import java.time.Instant; -import java.util.HashMap; -import java.util.Map; - -import static java.util.Objects.requireNonNull; - @Repository @RequiredArgsConstructor public class GithubIssueListRepoPg implements GithubIssueListRepo { @@ -141,7 +142,7 @@ public Single count(ListParams params) { return RxJava3Adapter.monoToSingle(spec.map(row -> row.get("total", Integer.class)).one()); } - private GithubIssueList.Item toGithubIssueListItem(Row row) { + private GithubIssueList.Item toGithubIssueListItem(Readable row) { String actionOwner = row.get("action_owner", String.class); if (actionOwner == null) { String[] moreActionOwners = requireNonNull(row.get("more_action_owners", String[].class)); @@ -166,9 +167,9 @@ private GithubIssueList.Item toGithubIssueListItem(Row row) { } @Override - public Flowable findAllActionOwner(ListParams params) { + public ImmutableList findAllActionOwner(ListParams params) { if (params.getActionOwner() != null) { - return Flowable.just(params.getActionOwner()); + return ImmutableList.of(params.getActionOwner()); } QuerySpec query = buildQuerySpec(params); @@ -185,20 +186,21 @@ public Flowable findAllActionOwner(ListParams params) { spec = spec.bind(binding.getKey(), binding.getValue()); } - return RxJava3Adapter.fluxToFlowable( - spec.map( - row -> { - String actionOwner = row.get("action_owner", String.class); - if (actionOwner == null) { - actionOwner = ""; - } - return actionOwner; - }) - .all()); + return spec.map( + row -> { + String actionOwner = row.get("action_owner", String.class); + if (actionOwner == null) { + actionOwner = ""; + } + return actionOwner; + }) + .all() + .collect(toImmutableList()) + .block(); } @Override - public Flowable findAllLabels(ListParams params) { + public ImmutableList findAllLabels(ListParams params) { QuerySpec query = buildQuerySpec(params); StringBuilder sql = new StringBuilder("SELECT DISTINCT unnest(gi.labels) AS label"); sql.append(query.from); @@ -209,7 +211,9 @@ public Flowable findAllLabels(ListParams params) { spec = spec.bind(binding.getKey(), binding.getValue()); } - return RxJava3Adapter.fluxToFlowable( - spec.map(row -> requireNonNull(row.get("label", String.class))).all()); + return spec.map(row -> requireNonNull(row.get("label", String.class))) + .all() + .collect(toImmutableList()) + .block(); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRestController.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRestController.java index f69984cb25..e4e96be7e2 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRestController.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListRestController.java @@ -1,15 +1,16 @@ package build.bazel.dashboard.github.issuelist; +import static build.bazel.dashboard.utils.RxJavaVirtualThread.single; +import static com.google.common.collect.ImmutableList.toImmutableList; + import build.bazel.dashboard.github.issuelist.GithubIssueListService.ListParams; import io.reactivex.rxjava3.core.Single; +import java.util.List; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; -import java.util.List; -import java.util.stream.Collectors; - @RestController @Slf4j @RequiredArgsConstructor @@ -23,19 +24,21 @@ public Single find(ListParams params) { @GetMapping("/github/issues/owners") public Single> findAllActionOwners(ListParams params) { - return githubIssueListService - .findAllActionOwner(params) - .filter(actionOwner -> !actionOwner.isBlank()) - .sorted() - .collect(Collectors.toList()); + return single( + () -> + githubIssueListService.findAllActionOwner(params).stream() + .filter(actionOwner -> !actionOwner.isBlank()) + .sorted() + .collect(toImmutableList())); } @GetMapping("/github/issues/labels") public Single> findAllLabels(ListParams params) { - return githubIssueListService - .findAllLabels(params) - .filter(label -> !label.isBlank()) - .sorted() - .collect(Collectors.toList()); + return single( + () -> + githubIssueListService.findAllLabels(params).stream() + .filter(label -> !label.isBlank()) + .sorted() + .collect(toImmutableList())); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListService.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListService.java index ec947436b3..de2c97acb1 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListService.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuelist/GithubIssueListService.java @@ -1,17 +1,19 @@ package build.bazel.dashboard.github.issuelist; +import static java.util.Objects.requireNonNull; + import build.bazel.dashboard.github.issuestatus.GithubIssueStatus; +import com.google.common.collect.ImmutableList; import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.core.Single; -import lombok.*; -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Service; - -import javax.annotation.Nullable; import java.util.List; import java.util.stream.Collectors; - -import static java.util.Objects.requireNonNull; +import javax.annotation.Nullable; +import lombok.Data; +import lombok.NoArgsConstructor; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; @Service @Slf4j @@ -72,12 +74,12 @@ public Single find(ListParams params) { .build())); } - public Flowable findAllActionOwner(ListParams params) { + public ImmutableList findAllActionOwner(ListParams params) { preprocessParams(params); return githubIssueListRepo.findAllActionOwner(params); } - public Flowable findAllLabels(ListParams params) { + public ImmutableList findAllLabels(ListParams params) { preprocessParams(params); return githubIssueListRepo.findAllLabels(params); } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutor.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutor.java index f39f31b8be..e8ba2a65b7 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutor.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutor.java @@ -15,9 +15,9 @@ class QueryResult { int totalCount; } - Single execute(String owner, String repo, String query); + QueryResult execute(String owner, String repo, String query); - default Single fetchQueryResultCount(String owner, String repo, String query) { - return execute(owner, repo, query).map(result -> result.totalCount); + default Integer fetchQueryResultCount(String owner, String repo, String query) { + return execute(owner, repo, query).totalCount; } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutorApi.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutorApi.java index 05765d45c0..341fbe5963 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutorApi.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutorApi.java @@ -15,22 +15,17 @@ public class GithubIssueQueryExecutorApi implements GithubIssueQueryExecutor { private final GithubApi githubApi; @Override - public Single execute(String owner, String repo, String query) { + public QueryResult execute(String owner, String repo, String query) { SearchIssuesRequest request = SearchIssuesRequest.builder().q(String.format("repo:%s/%s %s", owner, repo, query)).build(); - return githubApi - .searchIssues(request) - .flatMap( - response -> { - if (response.getStatus().is2xxSuccessful()) { - return Single.just( - QueryResult.builder() - .items(ImmutableList.copyOf(response.getBody().get("items"))) - .totalCount(response.getBody().get("total_count").asInt()) - .build()); - } else { - return Single.error(new IOException(response.getStatus().toString())); - } - }); + var response = githubApi.searchIssues(request); + if (response.getStatus().is2xxSuccessful()) { + return QueryResult.builder() + .items(ImmutableList.copyOf(response.getBody().get("items"))) + .totalCount(response.getBody().get("total_count").asInt()) + .build(); + } else { + throw new RuntimeException(response.getStatus().toString()); + } } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutorPg.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutorPg.java index c99759f18d..9c0a56a65d 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutorPg.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryExecutorPg.java @@ -3,6 +3,7 @@ import build.bazel.dashboard.github.issuequery.GithubIssueQueryParser.Query; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.collect.ImmutableList; import io.r2dbc.postgresql.codec.Json; import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.core.Single; @@ -19,6 +20,7 @@ import java.util.Map; import java.util.stream.Collectors; +import static com.google.common.collect.ImmutableList.toImmutableList; import static java.util.Objects.requireNonNull; @Slf4j @@ -30,16 +32,13 @@ public class GithubIssueQueryExecutorPg implements GithubIssueQueryExecutor { private final DatabaseClient databaseClient; @Override - public Single execute(String owner, String repo, String query) { - return fetchQueryResult(owner, repo, query) - .collect(Collectors.toList()) - .flatMap( - items -> - fetchQueryResultCount(owner, repo, query) - .map(count -> QueryResult.builder().totalCount(count).items(items).build())); + public QueryResult execute(String owner, String repo, String query) { + var items = fetchQueryResult(owner, repo, query); + var count = fetchQueryResultCount(owner, repo, query); + return QueryResult.builder().totalCount(count).items(items).build(); } - private Flowable fetchQueryResult(String owner, String repo, String query) { + private ImmutableList fetchQueryResult(String owner, String repo, String query) { SqlCondition condition = buildSqlCondition(owner, repo, query); StringBuilder sql = new StringBuilder( @@ -56,23 +55,22 @@ private Flowable fetchQueryResult(String owner, String repo, String qu spec = spec.bind(entry.getKey(), entry.getValue()); } - Flux result = - spec.map( - row -> { - try { - return objectMapper.readTree( - (requireNonNull(row.get("data", Json.class))).asArray()); - } catch (IOException e) { - throw new IllegalStateException(e); - } - }) - .all(); - - return RxJava3Adapter.fluxToFlowable(result); + return spec.map( + row -> { + try { + return objectMapper.readTree( + (requireNonNull(row.get("data", Json.class))).asArray()); + } catch (IOException e) { + throw new IllegalStateException(e); + } + }) + .all() + .collect(toImmutableList()) + .block(); } @Override - public Single fetchQueryResultCount(String owner, String repo, String query) { + public Integer fetchQueryResultCount(String owner, String repo, String query) { SqlCondition condition = buildSqlCondition(owner, repo, query); StringBuilder sql = new StringBuilder("SELECT COUNT(*) AS count FROM github_issue"); if (condition.condition.length() > 0) { @@ -85,7 +83,7 @@ public Single fetchQueryResultCount(String owner, String repo, String q spec = spec.bind(entry.getKey(), entry.getValue()); } - return RxJava3Adapter.monoToSingle(spec.map(row -> row.get("count", Integer.class)).one()); + return spec.map(row -> row.get("count", Integer.class)).one().block(); } private void and(StringBuilder conditions, String condition) { diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRepo.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRepo.java index fee95b6a3c..4ebbe2fa43 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRepo.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRepo.java @@ -1,7 +1,7 @@ package build.bazel.dashboard.github.issuequery; -import io.reactivex.rxjava3.core.Maybe; +import java.util.Optional; public interface GithubIssueQueryRepo { - Maybe findOne(String owner, String repo, String id); + Optional findOne(String owner, String repo, String id); } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRepoPg.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRepoPg.java index 004d3c0de7..50ea42fbb0 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRepoPg.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRepoPg.java @@ -1,15 +1,10 @@ package build.bazel.dashboard.github.issuequery; -import build.bazel.dashboard.github.issuequery.GithubIssueQuery; -import build.bazel.dashboard.github.issuequery.GithubIssueQueryRepo; -import io.reactivex.rxjava3.core.Maybe; +import java.time.Instant; +import java.util.Optional; import lombok.RequiredArgsConstructor; import org.springframework.r2dbc.core.DatabaseClient; import org.springframework.stereotype.Repository; -import reactor.adapter.rxjava.RxJava3Adapter; -import reactor.core.publisher.Mono; - -import java.time.Instant; @Repository @RequiredArgsConstructor @@ -17,11 +12,12 @@ public class GithubIssueQueryRepoPg implements GithubIssueQueryRepo { private final DatabaseClient databaseClient; @Override - public Maybe findOne(String owner, String repo, String id) { - Mono query = + public Optional findOne(String owner, String repo, String id) { + return Optional.ofNullable( databaseClient .sql( - "SELECT owner, repo, id, created_at, updated_at, name, query FROM github_issue_query WHERE owner = :owner AND repo = :repo AND id = :id") + "SELECT owner, repo, id, created_at, updated_at, name, query FROM" + + " github_issue_query WHERE owner = :owner AND repo = :repo AND id = :id") .bind("owner", owner) .bind("repo", repo) .bind("id", id) @@ -36,7 +32,7 @@ public Maybe findOne(String owner, String repo, String id) { .name(row.get("name", String.class)) .query(row.get("query", String.class)) .build()) - .one(); - return RxJava3Adapter.monoToMaybe(query); + .one() + .block()); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRestController.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRestController.java index 6728dd2416..8940178f25 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRestController.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/GithubIssueQueryRestController.java @@ -1,5 +1,8 @@ package build.bazel.dashboard.github.issuequery; +import static build.bazel.dashboard.utils.RxJavaVirtualThread.maybe; +import static build.bazel.dashboard.utils.RxJavaVirtualThread.single; + import build.bazel.dashboard.github.issuequery.GithubIssueQueryExecutor.QueryResult; import io.reactivex.rxjava3.core.Maybe; import io.reactivex.rxjava3.core.Single; @@ -23,7 +26,7 @@ public Single search( @PathVariable("owner") String owner, @PathVariable("repo") String repo, @RequestParam(name = "q") String q) { - return githubIssueQueryExecutor.execute(owner, repo, q); + return single(() -> githubIssueQueryExecutor.execute(owner, repo, q)); } @GetMapping("/internal/github/{owner}/{repo}/search/{queryId}") @@ -31,8 +34,10 @@ public Maybe searchByQueryId( @PathVariable("owner") String owner, @PathVariable("repo") String repo, @PathVariable("queryId") String queryId) { - return githubIssueQueryRepo - .findOne(owner, repo, queryId) - .flatMapSingle(query -> githubIssueQueryExecutor.execute(owner, repo, query.getQuery())); + return maybe( + () -> + githubIssueQueryRepo + .findOne(owner, repo, queryId) + .map(query -> githubIssueQueryExecutor.execute(owner, repo, query.getQuery()))); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/CountGithubIssueQueryTask.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/CountGithubIssueQueryTask.java index c6a26a2a48..9e960d0764 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/CountGithubIssueQueryTask.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/CountGithubIssueQueryTask.java @@ -1,16 +1,17 @@ package build.bazel.dashboard.github.issuequery.task; +import static build.bazel.dashboard.utils.RxJavaVirtualThread.completable; + import build.bazel.dashboard.github.issuequery.GithubIssueQueryExecutor; import build.bazel.dashboard.utils.Period; import io.reactivex.rxjava3.core.Completable; +import java.time.Instant; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.web.bind.annotation.PutMapping; import org.springframework.web.bind.annotation.RestController; -import java.time.Instant; - @RestController @Slf4j @RequiredArgsConstructor @@ -25,18 +26,16 @@ public void countDaily() { @PutMapping("/internal/github/search/count/daily") public Completable startCountDaily() { - log.info("Counting Github daily issue queries at {}...", Instant.now()); - return githubIssueQueryCountTaskRepo - .list(Period.DAILY) - .flatMapCompletable( - task -> - githubIssueQueryExecutor - .fetchQueryResultCount(task.getOwner(), task.getRepo(), task.getQuery()) - .flatMapCompletable( - count -> - githubIssueQueryCountTaskRepo.saveResult( - task, Instant.now(), count)), - false, - 1); + return completable( + () -> { + log.info("Counting Github daily issue queries at {}...", Instant.now()); + var tasks = githubIssueQueryCountTaskRepo.list(Period.DAILY); + for (var task : tasks) { + var count = + githubIssueQueryExecutor.fetchQueryResultCount( + task.getOwner(), task.getRepo(), task.getQuery()); + githubIssueQueryCountTaskRepo.saveResult(task, Instant.now(), count); + } + }); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRepo.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRepo.java index 3c361492b8..302eb6c126 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRepo.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRepo.java @@ -1,18 +1,14 @@ package build.bazel.dashboard.github.issuequery.task; -import build.bazel.dashboard.github.issuequery.task.GithubIssueQueryCountTask; -import build.bazel.dashboard.github.issuequery.task.GithubIssueQueryCountTaskResult; import build.bazel.dashboard.utils.Period; -import io.reactivex.rxjava3.core.Completable; -import io.reactivex.rxjava3.core.Flowable; - +import com.google.common.collect.ImmutableList; import java.time.Instant; public interface GithubIssueQueryCountTaskRepo { - Flowable list(Period period); + ImmutableList list(Period period); - Completable saveResult(GithubIssueQueryCountTask task, Instant timestamp, int count); + void saveResult(GithubIssueQueryCountTask task, Instant timestamp, int count); - Flowable listResult( + ImmutableList listResult( String owner, String repo, String queryId, Period period, Instant from, Instant to); } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRepoPg.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRepoPg.java index cb57b1f24b..f08f13862d 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRepoPg.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRepoPg.java @@ -1,8 +1,12 @@ package build.bazel.dashboard.github.issuequery.task; +import static com.google.common.collect.ImmutableList.toImmutableList; + import build.bazel.dashboard.utils.Period; +import com.google.common.collect.ImmutableList; import io.reactivex.rxjava3.core.Completable; import io.reactivex.rxjava3.core.Flowable; +import java.time.Instant; import lombok.RequiredArgsConstructor; import org.springframework.r2dbc.core.DatabaseClient; import org.springframework.stereotype.Repository; @@ -10,8 +14,6 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import java.time.Instant; - @Repository @RequiredArgsConstructor public class GithubIssueQueryCountTaskRepoPg @@ -19,73 +21,74 @@ public class GithubIssueQueryCountTaskRepoPg private final DatabaseClient databaseClient; @Override - public Flowable list(Period period) { - Flux query = - databaseClient - .sql( - "SELECT giqct.owner, giqct.repo, giqct.query_id, giqct.period, giqct.created_at, giq.query " - + "FROM github_issue_query_count_task giqct " - + "INNER JOIN github_issue_query giq ON giqct.owner = giq.owner AND giqct.repo = giq.repo AND giqct.query_id = giq.id " - + "WHERE giqct.period = :period") - .bind("period", period.toString()) - .map( - row -> - GithubIssueQueryCountTask.builder() - .owner(row.get("owner", String.class)) - .repo(row.get("repo", String.class)) - .queryId(row.get("query_id", String.class)) - .period(Period.valueOf(row.get("period", String.class))) - .createdAt(row.get("created_at", Instant.class)) - .query(row.get("query", String.class)) - .build()) - .all(); - return RxJava3Adapter.fluxToFlowable(query); + public ImmutableList list(Period period) { + return databaseClient + .sql( + "SELECT giqct.owner, giqct.repo, giqct.query_id, giqct.period, giqct.created_at," + + " giq.query FROM github_issue_query_count_task giqct INNER JOIN" + + " github_issue_query giq ON giqct.owner = giq.owner AND giqct.repo = giq.repo AND" + + " giqct.query_id = giq.id WHERE giqct.period = :period") + .bind("period", period.toString()) + .map( + row -> + GithubIssueQueryCountTask.builder() + .owner(row.get("owner", String.class)) + .repo(row.get("repo", String.class)) + .queryId(row.get("query_id", String.class)) + .period(Period.valueOf(row.get("period", String.class))) + .createdAt(row.get("created_at", Instant.class)) + .query(row.get("query", String.class)) + .build()) + .all() + .collect(toImmutableList()) + .block(); } @Override - public Completable saveResult(GithubIssueQueryCountTask task, Instant timestamp, int count) { - Mono execution = - databaseClient - .sql( - "INSERT INTO github_issue_query_count_task_result (owner, repo, query_id, period, timestamp, count) " - + "VALUES (:owner, :repo, :query_id, :period, :timestamp, :count) " - + "ON CONFLICT (owner, repo, query_id, period, timestamp) DO UPDATE " - + "SET count = excluded.count") - .bind("owner", task.getOwner()) - .bind("repo", task.getRepo()) - .bind("query_id", task.getQueryId()) - .bind("period", task.getPeriod().toString()) - .bind("timestamp", task.getPeriod().truncate(timestamp)) - .bind("count", count) - .then(); - return RxJava3Adapter.monoToCompletable(execution); + public void saveResult(GithubIssueQueryCountTask task, Instant timestamp, int count) { + databaseClient + .sql( + "INSERT INTO github_issue_query_count_task_result (owner, repo, query_id, period," + + " timestamp, count) VALUES (:owner, :repo, :query_id, :period, :timestamp," + + " :count) ON CONFLICT (owner, repo, query_id, period, timestamp) DO UPDATE" + + " SET count = excluded.count") + .bind("owner", task.getOwner()) + .bind("repo", task.getRepo()) + .bind("query_id", task.getQueryId()) + .bind("period", task.getPeriod().toString()) + .bind("timestamp", task.getPeriod().truncate(timestamp)) + .bind("count", count) + .then() + .block(); } @Override - public Flowable listResult( + public ImmutableList listResult( String owner, String repo, String queryId, Period period, Instant from, Instant to) { - Flux query = - databaseClient - .sql( - "SELECT owner, repo, query_id, period, timestamp, count FROM github_issue_query_count_task_result " - + "WHERE owner = :owner AND repo = :repo AND query_id = :query_id AND period = :period AND timestamp >= :from AND timestamp <= :to") - .bind("owner", owner) - .bind("repo", repo) - .bind("query_id", queryId) - .bind("period", period.toString()) - .bind("from", period.truncate(from)) - .bind("to", period.truncate(to)) - .map( - row -> - GithubIssueQueryCountTaskResult.builder() - .owner(row.get("owner", String.class)) - .repo(row.get("repo", String.class)) - .queryId(row.get("query_id", String.class)) - .period(Period.valueOf(row.get("period", String.class))) - .timestamp(row.get("timestamp", Instant.class)) - .count(row.get("count", Integer.class)) - .build()) - .all(); - return RxJava3Adapter.fluxToFlowable(query); + return databaseClient + .sql( + "SELECT owner, repo, query_id, period, timestamp, count FROM" + + " github_issue_query_count_task_result WHERE owner = :owner AND repo = :repo" + + " AND query_id = :query_id AND period = :period AND timestamp >= :from AND" + + " timestamp <= :to") + .bind("owner", owner) + .bind("repo", repo) + .bind("query_id", queryId) + .bind("period", period.toString()) + .bind("from", period.truncate(from)) + .bind("to", period.truncate(to)) + .map( + row -> + GithubIssueQueryCountTaskResult.builder() + .owner(row.get("owner", String.class)) + .repo(row.get("repo", String.class)) + .queryId(row.get("query_id", String.class)) + .period(Period.valueOf(row.get("period", String.class))) + .timestamp(row.get("timestamp", Instant.class)) + .count(row.get("count", Integer.class)) + .build()) + .all() + .collect(toImmutableList()) + .block(); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRestController.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRestController.java index 4a5ac6820e..ba3d52164e 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRestController.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuequery/task/GithubIssueQueryCountTaskRestController.java @@ -1,11 +1,18 @@ package build.bazel.dashboard.github.issuequery.task; +import static com.google.common.collect.ImmutableMap.toImmutableMap; + import build.bazel.dashboard.github.GithubUtils; import build.bazel.dashboard.github.issuequery.GithubIssueQueryExecutor; import build.bazel.dashboard.github.issuequery.GithubIssueQueryRepo; import build.bazel.dashboard.utils.Period; +import build.bazel.dashboard.utils.RxJavaVirtualThread; import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.core.Maybe; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; import lombok.Builder; import lombok.RequiredArgsConstructor; import lombok.Value; @@ -14,11 +21,6 @@ import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; -import java.time.Instant; -import java.util.ArrayList; -import java.util.List; -import java.util.stream.Collectors; - @RestController @RequiredArgsConstructor public class GithubIssueQueryCountTaskRestController { @@ -49,56 +51,50 @@ public Maybe fetchCountResult( @PathVariable("queryId") String queryId, @RequestParam("period") Period period, @RequestParam(name = "amount", defaultValue = "30") Integer amount) { - Instant now = Instant.now(); - Instant to = period.prev(now, 1); - Instant from = period.prev(to, amount); + return RxJavaVirtualThread.maybe( + () -> { + Instant now = Instant.now(); + Instant to = period.prev(now, 1); + Instant from = period.prev(to, amount); + var queryOptional = githubIssueQueryRepo.findOne(owner, repo, queryId); + if (queryOptional.isEmpty()) { + return Optional.empty(); + } + + var query = queryOptional.get(); + var resultMap = + githubIssueQueryCountTaskRepo + .listResult(query.getOwner(), query.getRepo(), query.getId(), period, from, to) + .stream() + .collect(toImmutableMap(GithubIssueQueryCountTaskResult::getTimestamp, it -> it)); - return githubIssueQueryRepo - .findOne(owner, repo, queryId) - .flatMapSingle( - query -> - githubIssueQueryCountTaskRepo - .listResult(query.getOwner(), query.getRepo(), query.getId(), period, from, to) - .collect( - Collectors.toMap(GithubIssueQueryCountTaskResult::getTimestamp, it -> it)) - .flatMap( - resultMap -> { - Instant end = period.truncate(to); - List items = new ArrayList<>(); - for (Instant timestamp = period.truncate(from); - !timestamp.isAfter(end); - timestamp = period.next(timestamp)) { - Integer count = null; - GithubIssueQueryCountTaskResult result = resultMap.get(timestamp); - if (result != null) { - count = result.getCount(); - } - CountResultItem item = - CountResultItem.builder().timestamp(timestamp).count(count).build(); - items.add(item); - } + Instant end = period.truncate(to); + List items = new ArrayList<>(); + for (Instant timestamp = period.truncate(from); + !timestamp.isAfter(end); + timestamp = period.next(timestamp)) { + Integer count = null; + GithubIssueQueryCountTaskResult result = resultMap.get(timestamp); + if (result != null) { + count = result.getCount(); + } + CountResultItem item = + CountResultItem.builder().timestamp(timestamp).count(count).build(); + items.add(item); + } - return githubIssueQueryExecutor - .fetchQueryResultCount(owner, repo, query.getQuery()) - .map( - count -> { - items.add( - CountResultItem.builder() - .timestamp(now) - .count(count) - .build()); - return CountResult.builder() - .id(query.getId()) - .name(query.getName()) - .url( - GithubUtils.buildIssueQueryUrl( - query.getOwner(), - query.getRepo(), - query.getQuery())) - .items(items) - .build(); - }); - })); + var count = githubIssueQueryExecutor.fetchQueryResultCount(owner, repo, query.getQuery()); + items.add(CountResultItem.builder().timestamp(now).count(count).build()); + return Optional.of( + CountResult.builder() + .id(query.getId()) + .name(query.getName()) + .url( + GithubUtils.buildIssueQueryUrl( + query.getOwner(), query.getRepo(), query.getQuery())) + .items(items) + .build()); + }); } @GetMapping("/github/{owner}/{repo}/search/count") diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRepo.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRepo.java index 23d0c59bec..d8c5e5aedc 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRepo.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRepo.java @@ -1,10 +1,9 @@ package build.bazel.dashboard.github.issuestatus; -import io.reactivex.rxjava3.core.Completable; -import io.reactivex.rxjava3.core.Maybe; +import java.util.Optional; public interface GithubIssueStatusRepo { - Completable save(GithubIssueStatus status); + void save(GithubIssueStatus status); - Maybe findOne(String owner, String repo, int issueNumber); + Optional findOne(String owner, String repo, int issueNumber); } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRepoPg.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRepoPg.java index 5dace231d8..90eec70175 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRepoPg.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRepoPg.java @@ -1,24 +1,21 @@ package build.bazel.dashboard.github.issuestatus; +import static java.util.Objects.requireNonNull; + import com.google.common.collect.ImmutableList; -import io.r2dbc.spi.Row; +import io.r2dbc.spi.Readable; import io.reactivex.rxjava3.core.Completable; -import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.core.Maybe; +import java.time.Instant; import java.util.ArrayDeque; -import java.util.ArrayList; import java.util.Deque; +import java.util.Optional; import lombok.RequiredArgsConstructor; import org.springframework.r2dbc.core.DatabaseClient; import org.springframework.stereotype.Repository; import reactor.adapter.rxjava.RxJava3Adapter; -import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import java.time.Instant; - -import static java.util.Objects.requireNonNull; - @Repository @RequiredArgsConstructor public class GithubIssueStatusRepoPg implements GithubIssueStatusRepo { @@ -26,7 +23,7 @@ public class GithubIssueStatusRepoPg implements GithubIssueStatusRepo { private final DatabaseClient databaseClient; @Override - public Completable save(GithubIssueStatus status) { + public void save(GithubIssueStatus status) { Deque actionOwners = new ArrayDeque<>(status.getActionOwners()); String actionOwner = null; if (!actionOwners.isEmpty()) { @@ -77,12 +74,12 @@ public Completable save(GithubIssueStatus status) { spec = spec.bindNull("next_notify_at", Instant.class); } - return RxJava3Adapter.monoToCompletable(spec.then()); + spec.then().block(); } @Override - public Maybe findOne(String owner, String repo, int issueNumber) { - Mono query = + public Optional findOne(String owner, String repo, int issueNumber) { + return Optional.ofNullable( databaseClient .sql( "SELECT owner, repo, issue_number, status, action_owner, more_action_owners," @@ -93,11 +90,11 @@ public Maybe findOne(String owner, String repo, int issueNumb .bind("repo", repo) .bind("issue_number", issueNumber) .map(this::toGithubIssueStatus) - .one(); - return RxJava3Adapter.monoToMaybe(query); + .one() + .block()); } - private GithubIssueStatus toGithubIssueStatus(Row row) { + private GithubIssueStatus toGithubIssueStatus(Readable row) { ImmutableList.Builder actionOwners = new ImmutableList.Builder<>(); String actionOwner = row.get("action_owner", String.class); if (actionOwner != null && !actionOwner.isBlank()) { diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRestController.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRestController.java index da8caf6ecb..ab342ce4a2 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRestController.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusRestController.java @@ -1,8 +1,10 @@ package build.bazel.dashboard.github.issuestatus; +import static build.bazel.dashboard.utils.RxJavaVirtualThread.completable; + import build.bazel.dashboard.github.issue.GithubIssueService; import io.reactivex.rxjava3.core.Completable; -import io.reactivex.rxjava3.core.Flowable; +import java.time.Instant; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.web.bind.annotation.PathVariable; @@ -10,8 +12,6 @@ import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; -import java.time.Instant; - @RestController @Slf4j @RequiredArgsConstructor @@ -25,9 +25,14 @@ public Completable checkStatus( @PathVariable("repo") String repo, @RequestParam(name = "start") Integer start, @RequestParam(name = "count") Integer count) { - return Completable.fromPublisher( - Flowable.range(start, count) - .flatMapMaybe(number -> githubIssueService.findOne(owner, repo, number)) - .flatMapMaybe(issue -> githubIssueStatusService.check(issue, Instant.now()))); + return completable( + () -> { + for (var number = start; number < start + count; ++number) { + var issue = githubIssueService.findOne(owner, repo, number); + if (issue.isPresent()) { + githubIssueStatusService.check(issue.get(), Instant.now()); + } + } + }); } } diff --git a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusService.java b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusService.java index 2d34eb33c0..6abc854ac3 100644 --- a/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusService.java +++ b/dashboard/server/src/main/java/build/bazel/dashboard/github/issuestatus/GithubIssueStatusService.java @@ -16,8 +16,10 @@ import io.reactivex.rxjava3.core.Completable; import io.reactivex.rxjava3.core.Maybe; import io.reactivex.rxjava3.core.Single; +import java.io.IOException; import java.time.Instant; import java.util.List; +import java.util.Optional; import javax.annotation.Nullable; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -32,69 +34,67 @@ public class GithubIssueStatusService { private final GithubIssueStatusRepo githubIssueStatusRepo; private final GithubTeamService githubTeamService; - public Maybe findOne(String owner, String repo, int issueNumber) { + public Optional findOne(String owner, String repo, int issueNumber) { return githubIssueStatusRepo.findOne(owner, repo, issueNumber); } - public Maybe check(GithubIssue issue, Instant now) { + public Optional check(GithubIssue issue, Instant now) throws IOException { String owner = issue.getOwner(); String repo = issue.getRepo(); - return githubRepoService.findOne(owner, repo).flatMap(githubRepo -> githubIssueStatusRepo - .findOne(owner, repo, issue.getIssueNumber()) - .flatMapSingle(status -> buildStatus(githubRepo, status, issue, now)) - .switchIfEmpty(buildStatus(githubRepo, null, issue, now)) - .flatMapMaybe(status -> githubIssueStatusRepo.save(status).andThen(Maybe.just(status)))); + var githubRepoOptional = githubRepoService.findOne(owner, repo); + if (githubRepoOptional.isEmpty()) { + return Optional.empty(); + } + + var githubRepo = githubRepoOptional.get(); + var existingIssueStatus = githubIssueStatusRepo.findOne(owner, repo, issue.getIssueNumber()); + + var newIssueStatus = buildStatus(githubRepo, existingIssueStatus.orElse(null), issue, now); + githubIssueStatusRepo.save(newIssueStatus); + return Optional.of(newIssueStatus); } - public Completable markDeleted(String owner, String repo, int issueNumber) { - return githubIssueStatusRepo - .findOne(owner, repo, issueNumber) - .flatMapCompletable( - status -> { - status.status = Status.DELETED; - return githubIssueStatusRepo.save(status); - }); + public void markDeleted(String owner, String repo, int issueNumber) { + githubIssueStatusRepo.findOne(owner, repo, issueNumber).ifPresent(status -> { + status.status = Status.DELETED; + githubIssueStatusRepo.save(status); + }); } - private Single buildStatus( + private GithubIssueStatus buildStatus( GithubRepo repo, - @Nullable GithubIssueStatus oldStatus, GithubIssue issue, Instant now) { - return Single.fromCallable(() -> issue.parseData(objectMapper)) - .flatMap( - data -> { - Status newStatus = newStatus(repo, data); - Instant lastNotifiedAt = null; - - if (oldStatus != null) { - lastNotifiedAt = oldStatus.getLastNotifiedAt(); - } - - Instant expectedRespondAt = getExpectedRespondAt(data, newStatus); - Instant nextNotifyAt = expectedRespondAt; - - if (lastNotifiedAt != null - && nextNotifyAt != null - && lastNotifiedAt.isAfter(expectedRespondAt)) { - nextNotifyAt = lastNotifiedAt.plus(1, DAYS); - } - - GithubIssueStatusBuilder builder = - GithubIssueStatus.builder() - .owner(issue.getOwner()) - .repo(issue.getRepo()) - .issueNumber(issue.getIssueNumber()) - .status(newStatus) - .updatedAt(data.getUpdatedAt()) - .expectedRespondAt(expectedRespondAt) - .lastNotifiedAt(lastNotifiedAt) - .nextNotifyAt(nextNotifyAt) - .checkedAt(now); - - return findActionOwner(repo, issue, data, newStatus) - .map(builder::actionOwners) - .map(GithubIssueStatusBuilder::build); - }); + @Nullable GithubIssueStatus oldStatus, GithubIssue issue, Instant now) throws IOException { + var data = issue.parseData(objectMapper); + Status newStatus = newStatus(repo, data); + Instant lastNotifiedAt = null; + + if (oldStatus != null) { + lastNotifiedAt = oldStatus.getLastNotifiedAt(); + } + + Instant expectedRespondAt = getExpectedRespondAt(data, newStatus); + Instant nextNotifyAt = expectedRespondAt; + + if (lastNotifiedAt != null + && nextNotifyAt != null + && lastNotifiedAt.isAfter(expectedRespondAt)) { + nextNotifyAt = lastNotifiedAt.plus(1, DAYS); + } + + var actionOwners = findActionOwner(repo, issue, data, newStatus); + return GithubIssueStatus.builder() + .owner(issue.getOwner()) + .repo(issue.getRepo()) + .issueNumber(issue.getIssueNumber()) + .status(newStatus) + .actionOwners(actionOwners) + .updatedAt(data.getUpdatedAt()) + .expectedRespondAt(expectedRespondAt) + .lastNotifiedAt(lastNotifiedAt) + .nextNotifyAt(nextNotifyAt) + .checkedAt(now) + .build(); } static Status newStatus(GithubRepo repo, GithubIssue.Data data) { @@ -150,40 +150,39 @@ static Status newStatus(GithubRepo repo, GithubIssue.Data data) { return null; } - private Single> findActionOwner( + private ImmutableList findActionOwner( GithubRepo repo, GithubIssue issue, GithubIssue.Data data, Status status) { switch (status) { case TO_BE_REVIEWED: - return Single.just(ImmutableList.of()); + return ImmutableList.of(); case MORE_DATA_NEEDED: - return Single.just(ImmutableList.of(data.getUser().getLogin())); + return ImmutableList.of(data.getUser().getLogin()); case REVIEWED: case TRIAGED: User assignee = data.getAssignee(); if (assignee != null) { - return Single.just(ImmutableList.of(assignee.getLogin())); + return ImmutableList.of(assignee.getLogin()); } else { List