BEP: Positively identify the last message

Set an additional flag in the last BEP message to indicate the end
of the proto sequence. While the last event is implicit from the
child-relation (the sequence of protos is finished, if at least one
event is seen and all announced events have occurred in the stream),
an explicit marking of the last event simplifies processing.

Change-Id: I6554476f975dc9e52fdb27fb3221bd27f6097ed9
PiperOrigin-RevId: 161377092

4 files changed, 72 insertions, 4 deletions

diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/LastBuildEvent.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/LastBuildEvent.java
new file mode 100644
index 0000000000..767e9140ed
--- /dev/null
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/LastBuildEvent.java
@@ -0,0 +1,40 @@
+// Copyright 2017 The Bazel Authors. All rights reserved.
+//
+// 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.google.devtools.build.lib.buildeventstream;
+
+import java.util.Collection;
+
+/** Wrapper class for a build event marking it as the final event in the protocol. */
+public class LastBuildEvent implements BuildEvent {
+  private final BuildEvent event;
+
+  public LastBuildEvent(BuildEvent event) {
+    this.event = event;
+  }
+
+  public BuildEventId getEventId() {
+    return event.getEventId();
+  }
+
+  public Collection<BuildEventId> getChildrenEvents() {
+    return event.getChildrenEvents();
+  }
+
+  public BuildEventStreamProtos.BuildEvent asStreamProto(BuildEventConverters converters) {
+    return BuildEventStreamProtos.BuildEvent.newBuilder(event.asStreamProto(converters))
+        .setLastMessage(true)
+        .build();
+  }
+}
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/proto/build_event_stream.proto b/src/main/java/com/google/devtools/build/lib/buildeventstream/proto/build_event_stream.proto
index ee27e6bd42..200784b535 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/proto/build_event_stream.proto
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/proto/build_event_stream.proto
@@ -486,6 +486,7 @@ message BuildFinished {
 message BuildEvent {
   BuildEventId id = 1;
   repeated BuildEventId children = 2;
+  bool last_message = 20;
   oneof payload {
     Progress progress = 3;
     Aborted aborted = 4;
diff --git a/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java b/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java
index f32518a99d..1d4e55b55e 100644
--- a/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java
+++ b/src/main/java/com/google/devtools/build/lib/runtime/BuildEventStreamer.java
@@ -44,6 +44,7 @@ import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos.Bui
 import com.google.devtools.build.lib.buildeventstream.BuildEventTransport;
 import com.google.devtools.build.lib.buildeventstream.BuildEventTransportClosedEvent;
 import com.google.devtools.build.lib.buildeventstream.BuildEventWithOrderConstraint;
+import com.google.devtools.build.lib.buildeventstream.LastBuildEvent;
 import com.google.devtools.build.lib.buildeventstream.NullConfiguration;
 import com.google.devtools.build.lib.buildeventstream.ProgressEvent;
 import com.google.devtools.build.lib.buildtool.BuildRequest;
@@ -172,14 +173,22 @@ public class BuildEventStreamer implements EventHandler {
     BuildEvent linkEvent = null;
     BuildEventId id = event.getEventId();
     List<BuildEvent> flushEvents = null;
+    boolean lastEvent = false;
 
     synchronized (this) {
       if (announcedEvents == null) {
         announcedEvents = new HashSet<>();
+        // The very first event of a stream is implicitly announced by the convention that
+        // a complete stream has to have at least one entry. In this way we keep the invariant
+        // that the set of posted events is always a subset of the set of announced events.
+        announcedEvents.add(id);
         if (!event.getChildrenEvents().contains(ProgressEvent.INITIAL_PROGRESS_UPDATE)) {
           linkEvent = ProgressEvent.progressChainIn(progressCount, event.getEventId());
           progressCount++;
           announcedEvents.addAll(linkEvent.getChildrenEvents());
+          // the new first event in the stream, implicitly announced by the fact that complete
+          // stream may not be empty.
+          announcedEvents.add(linkEvent.getEventId());
           postedEvents.add(linkEvent.getEventId());
         }
 
@@ -219,13 +228,23 @@ public class BuildEventStreamer implements EventHandler {
 
       postedEvents.add(id);
       announcedEvents.addAll(event.getChildrenEvents());
+      // We keep as an invariant that postedEvents is a subset of announced events, so this is a
+      // cheaper test for equality
+      if (announcedEvents.size() == postedEvents.size()) {
+        lastEvent = true;
+      }
+    }
+
+    BuildEvent mainEvent = event;
+    if (lastEvent) {
+      mainEvent = new LastBuildEvent(event);
     }
 
     for (BuildEventTransport transport : transports) {
       if (linkEvent != null) {
         transport.sendBuildEvent(linkEvent, artifactGroupNamer);
       }
-      transport.sendBuildEvent(event, artifactGroupNamer);
+      transport.sendBuildEvent(mainEvent, artifactGroupNamer);
     }
 
     if (flushEvents != null) {
diff --git a/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java b/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java
index a3b5b7e4e1..cda02a7eff 100644
--- a/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java
+++ b/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java
@@ -269,7 +269,9 @@ public class BuildEventStreamerTest extends FoundationTestCase {
   @Test(timeout = 5000)
   public void testSimpleStream() {
     // Verify that a well-formed event is passed through and that completion of the
-    // build clears the pending progress-update event.
+    // build clears the pending progress-update event. However, there is no guarantee
+    // on the order of the flushed events.
+    // Additionally, assert that the actual last event has the last_message flag set.
 
     EventBusHandler handler = new EventBusHandler();
     eventBus.register(handler);
@@ -295,8 +297,14 @@ public class BuildEventStreamerTest extends FoundationTestCase {
 
     List<BuildEvent> finalStream = transport.getEvents();
     assertThat(finalStream).hasSize(3);
-    assertThat(finalStream.get(1).getEventId()).isEqualTo(BuildEventId.buildFinished());
-    assertThat(finalStream.get(2).getEventId()).isEqualTo(ProgressEvent.INITIAL_PROGRESS_UPDATE);
+    assertThat(ImmutableSet.of(finalStream.get(1).getEventId(), finalStream.get(2).getEventId()))
+        .isEqualTo(
+            ImmutableSet.of(BuildEventId.buildFinished(), ProgressEvent.INITIAL_PROGRESS_UPDATE));
+
+    // verify the "last_message" flag.
+    assertThat(transport.getEventProtos().get(0).getLastMessage()).isFalse();
+    assertThat(transport.getEventProtos().get(1).getLastMessage()).isFalse();
+    assertThat(transport.getEventProtos().get(2).getLastMessage()).isTrue();
 
     while (!handler.transportSet.isEmpty()) {
       LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100));