EventParsing: decrease heap impact Skip intermediate unprocessedCommits Collection. Introduce new UnprocessedCommitsWalker that provides iteration over unprocessed commits without extra heap-usage by using the RevWalk's internal data structure. Solves: Jira GER-1558 Change-Id: I35fbf4a79de4ae6f6208881a12931c66db3a926e
diff --git a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/parsing/EiffelEventParser.java b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/parsing/EiffelEventParser.java index 06057f8..799c476 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/parsing/EiffelEventParser.java +++ b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/parsing/EiffelEventParser.java
@@ -39,8 +39,8 @@ import com.googlesource.gerrit.plugins.eventseiffel.eiffel.SourceChangeEventKey; import com.googlesource.gerrit.plugins.eventseiffel.eiffel.dto.EiffelEvent; import com.googlesource.gerrit.plugins.eventseiffel.mapping.EiffelEventMapper; +import com.googlesource.gerrit.plugins.eventseiffel.parsing.UnprocessedCommitsWalker.EventCreate; import java.io.IOException; -import java.util.Deque; import java.util.List; import java.util.Optional; import java.util.UUID; @@ -55,9 +55,6 @@ import org.eclipse.jgit.lib.Ref; import org.eclipse.jgit.lib.Repository; import org.eclipse.jgit.revwalk.RevCommit; -import org.eclipse.jgit.revwalk.RevFlag; -import org.eclipse.jgit.revwalk.RevSort; -import org.eclipse.jgit.revwalk.RevWalk; /** Creates and pushes missing Eiffel events to the Eiffel event queue. */ public class EiffelEventParser { @@ -67,13 +64,18 @@ private final EiffelEventHub eventHub; private final GitRepositoryManager repoManager; + private final UnprocessedCommitsWalker.Factory walkerFactory; private final EiffelEventMapper mapper; @Inject public EiffelEventParser( - EiffelEventHub eventQueue, GitRepositoryManager repoManager, EiffelEventMapper mapper) { + EiffelEventHub eventQueue, + GitRepositoryManager repoManager, + EiffelEventMapper mapper, + UnprocessedCommitsWalker.Factory walkerFactory) { this.eventHub = eventQueue; this.repoManager = repoManager; + this.walkerFactory = walkerFactory; this.mapper = mapper; } @@ -97,7 +99,7 @@ } } - /* Eiffel events have been scheduled or created for all parents. */ + /* Eiffel events have been scheduled or published for all parents. */ if (parentUuids.size() == commit.parents.size()) { pushToHub(mapper.toScc(event, parentUuids)); } else { @@ -153,66 +155,55 @@ String commitSha1TransactionEnd, AccountInfo submitter, Long submittedAt) { - Deque<EventCreate> commitsToProcess = null; - try { - logger.atFine().log("Start parsing commits for: %s", scs); - commitsToProcess = - commitsUnprocessedForScs(scs, commitSha1TransactionEnd, submitter, submittedAt); - if (commitsToProcess.isEmpty()) { - logger.atInfo().log("All events were already created for %s", scs); - } - logger.atFine().log("Done parsing commits for: %s", scs); - } catch (IOException | EiffelEventIdLookupException e) { - logger.atSevere().withCause(e).log( - "Failed to get commits to process for %s, %s, %s.", - scs.repo(), scs.branch(), scs.commit()); - } - SourceChangeEventKey currentScs = scs; + SourceChangeEventKey scc = scs.copy(SCC); try { - SourceChangeEventKey scc = scs.copy(SCC); if (eventHub.getExistingId(scc).isEmpty()) { /* One or several SCC events are missing, create them first */ createAndScheduleMissingSccs(scc); } - - logger.atFine().log("Start publishing events for: %s", scs); - while (commitsToProcess != null && !commitsToProcess.isEmpty()) { - EventCreate create = commitsToProcess.pop(); - currentScs = create.key; - Optional<UUID> sccId = eventHub.getExistingId(create.key.copy(SCC)); - if (sccId.isEmpty()) { - throw new EiffelEventIdLookupException( - "Unable to find SCC event id: %s", create.key.copy(SCC)); + try (UnprocessedCommitsWalker commitFinder = + walkerFactory.scsWalker(scs, commitSha1TransactionEnd, submitter, submittedAt)) { + if (!commitFinder.hasNext()) { + logger.atInfo().log("All events were already published for %s", scs); + } else { + logger.atFine().log("Start publishing events for: %s", scs); } - List<UUID> scsParentEventIds = Lists.newArrayList(); - for (RevCommit parent : create.commit.getParents()) { - Optional<UUID> parentUuid = eventHub.getExistingId(create.key.copy(parent.getName())); - if (parentUuid.isPresent()) { - scsParentEventIds.add(parentUuid.get()); - } else { - exceptionForMissingParent(create, parent); + while (commitFinder.hasNext()) { + EventCreate create = commitFinder.next(); + currentScs = create.key; + Optional<UUID> sccId = eventHub.getExistingId(create.key.copy(SCC)); + if (sccId.isEmpty()) { + throw new EiffelEventIdLookupException( + "Unable to find SCC event id: %s", create.key.copy(SCC)); } + List<UUID> scsParentEventIds = Lists.newArrayList(); + for (RevCommit parent : create.commit.getParents()) { + Optional<UUID> parentUuid = eventHub.getExistingId(create.key.copy(parent.getName())); + if (parentUuid.isPresent()) { + scsParentEventIds.add(parentUuid.get()); + } else { + exceptionForMissingParent(create, parent); + } + } + pushToHub( + mapper.toScs( + create.commit, + create.key.repo(), + create.key.branch(), + create.submitter, + create.submittedAt, + scsParentEventIds, + sccId.get())); } - pushToHub( - mapper.toScs( - create.commit, - create.key.repo(), - create.key.branch(), - create.submitter, - create.submittedAt, - scsParentEventIds, - sccId.get())); } logger.atFine().log("Done publishing events for: %s", scs); - } catch (EiffelEventIdLookupException - | IOException - | NoSuchEntityException + } catch (IOException + | EiffelEventIdLookupException + | InterruptedException | ConfigInvalidException - | InterruptedException e) { - logger.atSevere().withCause(e).log( - "Failed to create Eiffel event(s) for %s skipping another %s dependent events.", - currentScs, commitsToProcess.size()); + | NoSuchEntityException e) { + logger.atSevere().withCause(e).log("Failed to create Eiffel event(s) for %s.", currentScs); } } @@ -317,139 +308,37 @@ throws MissingObjectException, IncorrectObjectTypeException, IOException, EiffelEventIdLookupException, RepositoryNotFoundException, NoSuchEntityException, ConfigInvalidException, InterruptedException { - logger.atFine().log("Start parsing commits for: %s", scc); - Deque<EventCreate> unprocessedCommits = commitsUnprocessedForScc(scc); - if (unprocessedCommits.isEmpty()) { - logger.atInfo().log("All events were already created for %s", scc); - } - logger.atFine().log("Done parsing commits for: %s", scc); logger.atFine().log("Start publishing events for: %s", scc); - while (!unprocessedCommits.isEmpty()) { - EventCreate job = unprocessedCommits.pop(); - logger.atFine().log("Processing event-creation for: %s", job.key); - List<UUID> parentIds = Lists.newArrayList(); - for (RevCommit parent : job.commit.getParents()) { - SourceChangeEventKey parentKey = scc.copy(parent.getName()); - Optional<UUID> parentId = eventHub.getExistingId(parentKey); - if (parentId.isPresent()) { - parentIds.add(parentId.get()); - } else { - exceptionForMissingParent(job, parent); - } + try (UnprocessedCommitsWalker commitFinder = walkerFactory.sccWalker(scc)) { + if (!commitFinder.hasNext()) { + logger.atInfo().log("All events were already published for %s", scc); + } else { + logger.atFine().log("Start publishing events for: %s", scc); } - try { - pushToHub(mapper.toScc(job.commit, job.key.repo(), job.key.branch(), parentIds)); - } catch (InterruptedException e) { - logger.atSevere().log( - "Interrupted while pushing %s to EventHub with %d events left to process", - job.key, unprocessedCommits.size()); - throw e; + while (commitFinder.hasNext()) { + EventCreate job = commitFinder.next(); + logger.atFine().log("Processing event-creation for: %s", job.key); + List<UUID> parentIds = Lists.newArrayList(); + for (RevCommit parent : job.commit.getParents()) { + SourceChangeEventKey parentKey = scc.copy(parent.getName()); + Optional<UUID> parentId = eventHub.getExistingId(parentKey); + if (parentId.isPresent()) { + parentIds.add(parentId.get()); + } else { + exceptionForMissingParent(job, parent); + } + } + try { + pushToHub(mapper.toScc(job.commit, job.key.repo(), job.key.branch(), parentIds)); + } catch (InterruptedException e) { + logger.atSevere().log("Interrupted while pushing %s to EventHub.", job.key); + throw e; + } } } logger.atFine().log("Done publishing events for: %s", scc); } - private Deque<EventCreate> commitsUnprocessedForScc(SourceChangeEventKey key) - throws MissingObjectException, IncorrectObjectTypeException, IOException, - EiffelEventIdLookupException, RepositoryNotFoundException { - Deque<EventCreate> unprocessedCommits = Lists.newLinkedList(); - try (Repository repo = repoManager.openRepository(Project.nameKey(key.repo())); - RevWalk rw = new RevWalk(repo)) { - rw.setRetainBody(false); - RevFlag hasSccEventFlag = rw.newFlag("HAS_SCC_EVENT"); - boolean hasEvents = hasEvents(key, rw); - rw.markStart(rw.parseCommit(ObjectId.fromString(key.commit()))); - rw.sort(RevSort.TOPO); - for (RevCommit commit : rw) { - SourceChangeEventKey current = key.copy(commit.getName()); - /* If branch is previously unhandled there's no point in asking eventHub/Event Repository - * since no events will be found. */ - if (hasEvents && isAlreadyHandled(hasSccEventFlag, commit, current)) { - logger.atFine().log("Event already created for %s", current); - flagWithParents(commit, hasSccEventFlag); - } else { - rw.parseBody(commit); - unprocessedCommits.push(new EventCreate(current, commit)); - logger.atFine().log("Added unhandled event: %s, to queue", current); - } - } - } - return unprocessedCommits; - } - - private Deque<EventCreate> commitsUnprocessedForScs( - SourceChangeEventKey scs, - String commitSha1TransactionEnd, - AccountInfo submitter, - Long submittedAt) - throws IOException, EiffelEventIdLookupException { - Deque<EventCreate> unhandledScss = Lists.newLinkedList(); - try (Repository repo = repoManager.openRepository(Project.nameKey(scs.repo())); - RevWalk rw = new RevWalk(repo)) { - rw.setRetainBody(false); - RevFlag hasScsEventFlag = rw.newFlag("HAS_SCS_EVENT"); - boolean hasEvents = hasEvents(scs, rw); - if (commitSha1TransactionEnd != null - && !commitSha1TransactionEnd.equals(ObjectId.zeroId().name())) { - rw.markUninteresting(rw.parseCommit(ObjectId.fromString(commitSha1TransactionEnd))); - } - rw.markStart(rw.parseCommit(ObjectId.fromString(scs.commit()))); - rw.sort(RevSort.TOPO); - for (RevCommit commit : rw) { - SourceChangeEventKey current = scs.copy(commit); - /* If branch is previously unhandled there's no point in asking eventHub/Event Repository - * since no events will be found. */ - if (hasEvents && isAlreadyHandled(hasScsEventFlag, commit, current)) { - flagWithParents(commit, hasScsEventFlag); - } else { - rw.parseBody(commit); - unhandledScss.push(new EventCreate(current, commit, submitter, submittedAt)); - } - } - if (commitSha1TransactionEnd != null - && !commitSha1TransactionEnd.equals(ObjectId.zeroId().name()) - && eventHub.getExistingId(scs.copy(commitSha1TransactionEnd)).isEmpty()) { - rw.reset(); - rw.markStart(rw.parseCommit(ObjectId.fromString(commitSha1TransactionEnd))); - rw.sort(RevSort.TOPO); - for (RevCommit commit : rw) { - SourceChangeEventKey current = scs.copy(commit); - /* If branch is previously unhandled there's no point in asking eventHub/Event Repository - * since no events will be found. */ - if (hasEvents && eventHub.getExistingId(current).isPresent()) { - rw.markUninteresting(commit); - } else { - rw.parseBody(commit); - unhandledScss.push(new EventCreate(current, rw.parseCommit(commit))); - } - } - } - } - return unhandledScss; - } - - /* Returns true if an event has been created for key.branch from any commits reachable from - * key.commit (i.e. initial commit) */ - private boolean hasEvents(SourceChangeEventKey key, RevWalk rw) - throws MissingObjectException, IncorrectObjectTypeException, IOException, - EiffelEventIdLookupException { - rw.sort(RevSort.REVERSE); - rw.markStart(rw.parseCommit(ObjectId.fromString(key.commit()))); - RevCommit initialCommit = rw.next(); - if (initialCommit == null) { - logger.atWarning().log("Unable to find initial commit for: %s", key); - return false; - } - logger.atFine().log("Found initial commit: \"%s\" of %s ", initialCommit.name(), key); - - /* Reset RevWalk. */ - rw.reset(); - - Optional<UUID> eventId = eventHub.getExistingId(key.copy(initialCommit)); - eventId.ifPresent(id -> logger.atFine().log("%s has events", key)); - return eventId.isPresent(); - } - private ObjectId getTipOf(String repoName, String branch) { try (Repository repo = repoManager.openRepository(Project.nameKey(repoName))) { Ref branchRef = repo.exactRef(branch); @@ -471,36 +360,4 @@ "Unable to lookup parent (%s) event UUID for %s even though it should exist.", parent.abbreviate(7).name(), create.key)); } - - private static void flagWithParents(RevCommit commit, RevFlag flag) { - commit.add(flag); - for (RevCommit parent : commit.getParents()) { - parent.add(flag); - } - } - - private boolean isAlreadyHandled(RevFlag hasEventFlag, RevCommit commit, SourceChangeEventKey key) - throws EiffelEventIdLookupException { - logger.atFine().log("%s has event flag: %b", key.copy(commit), commit.has(hasEventFlag)); - return commit.has(hasEventFlag) || eventHub.getExistingId(key).isPresent(); - } - - private static class EventCreate { - SourceChangeEventKey key; - RevCommit commit; - AccountInfo submitter; - Long submittedAt; - - public EventCreate(SourceChangeEventKey key, RevCommit commit) { - this(key, commit, null, null); - } - - public EventCreate( - SourceChangeEventKey key, RevCommit commit, AccountInfo submitter, Long submittedAt) { - this.key = key; - this.commit = commit; - this.submitter = submitter; - this.submittedAt = submittedAt; - } - } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/parsing/UnprocessedCommitsWalker.java b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/parsing/UnprocessedCommitsWalker.java new file mode 100644 index 0000000..6cf850d --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/eventseiffel/parsing/UnprocessedCommitsWalker.java
@@ -0,0 +1,303 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.eventseiffel.parsing; + +import static com.google.common.base.Preconditions.checkState; +import static com.googlesource.gerrit.plugins.eventseiffel.eiffel.dto.EiffelEventType.SCC; +import static com.googlesource.gerrit.plugins.eventseiffel.eiffel.dto.EiffelEventType.SCS; + +import com.google.common.flogger.FluentLogger; +import com.google.gerrit.entities.Project; +import com.google.gerrit.extensions.common.AccountInfo; +import com.google.gerrit.server.git.GitRepositoryManager; +import com.google.inject.Inject; +import com.google.inject.Singleton; +import com.googlesource.gerrit.plugins.eventseiffel.EiffelEventHub; +import com.googlesource.gerrit.plugins.eventseiffel.cache.EiffelEventIdLookupException; +import com.googlesource.gerrit.plugins.eventseiffel.eiffel.SourceChangeEventKey; +import java.io.IOException; +import java.util.Optional; +import java.util.UUID; +import org.eclipse.jgit.errors.IncorrectObjectTypeException; +import org.eclipse.jgit.errors.MissingObjectException; +import org.eclipse.jgit.errors.RepositoryNotFoundException; +import org.eclipse.jgit.lib.ObjectId; +import org.eclipse.jgit.lib.Repository; +import org.eclipse.jgit.revwalk.RevCommit; +import org.eclipse.jgit.revwalk.RevFlag; +import org.eclipse.jgit.revwalk.RevSort; +import org.eclipse.jgit.revwalk.RevWalk; + +public abstract class UnprocessedCommitsWalker implements AutoCloseable { + @Singleton + public static class Factory { + private final EiffelEventHub eventHub; + private final GitRepositoryManager repoManager; + + @Inject + public Factory(EiffelEventHub eventHub, GitRepositoryManager repoManager) { + this.eventHub = eventHub; + this.repoManager = repoManager; + } + + public UnprocessedCommitsWalker sccWalker(SourceChangeEventKey eventKey) + throws RepositoryNotFoundException, IOException, EiffelEventIdLookupException { + return new SccWalker(eventKey, eventHub, repoManager); + } + + public UnprocessedCommitsWalker scsWalker( + SourceChangeEventKey eventKey, + String commitSha1TransactionEnd, + AccountInfo submitter, + Long submittedAt) + throws RepositoryNotFoundException, IOException, EiffelEventIdLookupException { + return new ScsWalker( + eventKey, eventHub, repoManager, commitSha1TransactionEnd, submitter, submittedAt); + } + } + + private static final FluentLogger logger = FluentLogger.forEnclosingClass(); + + private static void flagWithParents(RevCommit commit, RevFlag flag) { + commit.add(flag); + for (RevCommit parent : commit.getParents()) { + parent.add(flag); + } + } + + protected final SourceChangeEventKey eventKey; + protected final EiffelEventHub eventHub; + protected final Repository repo; + protected final RevWalk rw; + protected final boolean hasEvents; + protected EventCreate next; + + UnprocessedCommitsWalker( + SourceChangeEventKey eventKey, EiffelEventHub eventHub, GitRepositoryManager repoManager) + throws RepositoryNotFoundException, IOException, EiffelEventIdLookupException { + this.eventKey = eventKey; + this.eventHub = eventHub; + this.repo = repoManager.openRepository(Project.nameKey(eventKey.repo())); + this.rw = new RevWalk(repo); + rw.setRetainBody(false); + this.hasEvents = hasEvents(); + } + + @Override + public void close() throws IOException { + rw.close(); + repo.close(); + } + + boolean hasNext() { + return next != null; + } + + EventCreate next() throws MissingObjectException, EiffelEventIdLookupException, IOException { + EventCreate current = next; + setNext(); + return current; + } + + protected abstract void setNext() + throws MissingObjectException, EiffelEventIdLookupException, IOException; + + protected SourceChangeEventKey toKey(RevCommit commit) { + return eventKey.copy(commit.getName()); + } + + protected boolean isAlreadyHandled( + RevFlag hasEventFlag, RevCommit commit, SourceChangeEventKey key) + throws EiffelEventIdLookupException { + logger.atFine().log("%s has event flag: %b", key.copy(commit), commit.has(hasEventFlag)); + return commit.has(hasEventFlag) || eventHub.getExistingId(key).isPresent(); + } + + /* Returns true if an event has been created for key.branch from any commits reachable from + * key.commit (i.e. initial commit) */ + private boolean hasEvents() + throws MissingObjectException, IncorrectObjectTypeException, IOException, + EiffelEventIdLookupException { + rw.sort(RevSort.REVERSE); + rw.markStart(rw.parseCommit(ObjectId.fromString(eventKey.commit()))); + RevCommit initialCommit = rw.next(); + if (initialCommit == null) { + logger.atWarning().log("Unable to find initial commit for: %s", eventKey); + return false; + } + logger.atFine().log("Found initial commit: \"%s\" of %s.", initialCommit.name(), eventKey); + + /* Reset RevWalk. */ + rw.reset(); + + Optional<UUID> eventId = eventHub.getExistingId(toKey(initialCommit)); + eventId.ifPresent(id -> logger.atFine().log("%s has events", eventKey)); + return eventId.isPresent(); + } + + public class EventCreate { + SourceChangeEventKey key; + RevCommit commit; + AccountInfo submitter; + Long submittedAt; + + public EventCreate(RevCommit commit) throws MissingObjectException, IOException { + this(commit, null, null); + } + + public EventCreate(RevCommit commit, AccountInfo submitter, Long submittedAt) + throws MissingObjectException, IOException { + rw.parseBody(commit); + this.key = toKey(commit); + this.commit = commit; + this.submitter = submitter; + this.submittedAt = submittedAt; + } + } + + public static class SccWalker extends UnprocessedCommitsWalker { + private RevFlag hasSccEventFlag; + + SccWalker( + SourceChangeEventKey eventKey, EiffelEventHub eventHub, GitRepositoryManager repoManager) + throws RepositoryNotFoundException, IOException, EiffelEventIdLookupException { + super(eventKey, eventHub, repoManager); + checkState(eventKey.type().equals(SCC), "EventKey must have type SCC for SccWalker."); + this.hasSccEventFlag = rw.newFlag("HAS_SCC_EVENT"); + parse(); + setNext(); + } + + protected void parse() + throws MissingObjectException, EiffelEventIdLookupException, IOException { + rw.markStart(rw.parseCommit(ObjectId.fromString(eventKey.commit()))); + rw.sort(RevSort.TOPO); + RevCommit commit; + while ((commit = rw.next()) != null) { + SourceChangeEventKey current = toKey(commit); + /* If branch is previously unhandled there's no point in asking eventHub/Event Repository + * since no events will be found. */ + if (hasEvents && isAlreadyHandled(hasSccEventFlag, commit, current)) { + logger.atFine().log("Event already created for %s", current); + flagWithParents(commit, hasSccEventFlag); + rw.markUninteresting(commit); + } + } + rw.resetRetain(RevFlag.UNINTERESTING, hasSccEventFlag); + rw.sort(RevSort.TOPO, true); + rw.sort(RevSort.REVERSE, true); + rw.markStart(rw.parseCommit(ObjectId.fromString(eventKey.commit()))); + } + + @Override + protected void setNext() + throws MissingObjectException, EiffelEventIdLookupException, IOException { + next = null; + RevCommit commit = rw.next(); + if (commit == null) { + return; + } + next = new EventCreate(commit); + } + } + + public static class ScsWalker extends UnprocessedCommitsWalker { + + private final RevCommit transactionEnd; + private final AccountInfo submitter; + private final Long submittedAt; + private RevFlag hasScsEventFlag; + private RevFlag outsideTransaction; + + ScsWalker( + SourceChangeEventKey eventKey, + EiffelEventHub eventHub, + GitRepositoryManager repoManager, + String commitSha1TransactionEnd, + AccountInfo submitter, + Long submittedAt) + throws RepositoryNotFoundException, IOException, EiffelEventIdLookupException { + super(eventKey, eventHub, repoManager); + checkState(eventKey.type().equals(SCS), "EventKey must have type SCS for ScsWalker."); + this.transactionEnd = + commitSha1TransactionEnd != null + ? rw.parseCommit(ObjectId.fromString(commitSha1TransactionEnd)) + : null; + this.submitter = submitter; + this.submittedAt = submittedAt; + this.hasScsEventFlag = rw.newFlag("HAS_SCS_EVENT"); + this.outsideTransaction = rw.newFlag("OUTSIDE_TRANSACTION"); + parse(); + setNext(); + } + + protected void parse() + throws MissingObjectException, EiffelEventIdLookupException, IOException { + rw.markStart(rw.parseCommit(ObjectId.fromString(eventKey.commit()))); + rw.sort(RevSort.TOPO); + if (transactionEnd != null && !transactionEnd.equals(ObjectId.zeroId())) { + rw.markUninteresting(transactionEnd); + } + RevCommit commit; + while ((commit = rw.next()) != null) { + SourceChangeEventKey current = toKey(commit); + /* If branch is previously unhandled there's no point in asking eventHub/Event Repository + * since no events will be found. */ + if (hasEvents && isAlreadyHandled(hasScsEventFlag, commit, current)) { + flagWithParents(commit, hasScsEventFlag); + rw.markUninteresting(commit); + } + } + if (transactionEnd != null + && !transactionEnd.equals(ObjectId.zeroId()) + && eventHub.getExistingId(eventKey.copy(transactionEnd)).isEmpty()) { + rw.resetRetain(RevFlag.UNINTERESTING, hasScsEventFlag); + transactionEnd.remove(RevFlag.UNINTERESTING); + rw.markStart(transactionEnd); + rw.sort(RevSort.TOPO); + while ((commit = rw.next()) != null) { + SourceChangeEventKey current = eventKey.copy(commit); + /* If branch is previously unhandled there's no point in asking eventHub/Event Repository + * since no events will be found. */ + if (hasEvents && isAlreadyHandled(hasScsEventFlag, commit, current)) { + flagWithParents(commit, hasScsEventFlag); + rw.markUninteresting(commit); + } else { + commit.add(outsideTransaction); + } + } + } + rw.resetRetain(RevFlag.UNINTERESTING, hasScsEventFlag, outsideTransaction); + rw.sort(RevSort.TOPO, true); + rw.sort(RevSort.REVERSE, true); + rw.markStart(rw.parseCommit(ObjectId.fromString(eventKey.commit()))); + } + + @Override + protected void setNext() + throws MissingObjectException, EiffelEventIdLookupException, IOException { + next = null; + RevCommit commit = rw.next(); + if (commit == null) { + return; + } + if (commit.has(outsideTransaction)) { + next = new EventCreate(commit); + } else { + next = new EventCreate(commit, submitter, submittedAt); + } + } + } +}