Enable the "received byte count" feature.

library/common/extensions/filters/http/platform_bridge/filter.cc

@@ -202,7 +202,7 @@ envoy_stream_intel PlatformBridgeFilter::streamIntel() {
   RELEASE_ASSERT(decoder_callbacks_, "StreamInfo accessed before filter callbacks are set");
   auto& info = decoder_callbacks_->streamInfo();
   // FIXME: Stream handle cannot currently be set from the filter context.
-  envoy_stream_intel stream_intel{-1, -1, 0};
+  envoy_stream_intel stream_intel{-1, -1, 0, 0};
   if (info.upstreamInfo()) {
     stream_intel.connection_id = info.upstreamInfo()->upstreamConnectionId().value_or(-1);
   }

library/common/http/client.cc

@@ -298,6 +298,9 @@ void Client::DirectStream::saveLatestStreamIntel() {
   }
   stream_intel_.stream_id = static_cast<uint64_t>(stream_handle_);
   stream_intel_.attempt_count = info.attemptCount().value_or(0);
+  if (info.getUpstreamBytesMeter()) {
+    stream_intel_.received_byte_count = info.getUpstreamBytesMeter()->wireBytesReceived();
+  }
 }
 
 void Client::DirectStream::saveFinalStreamIntel() {

library/common/http/client.h

@@ -279,7 +279,7 @@ class Client : public Logger::Loggable<Logger::Id::http> {
     // read faster than the mobile caller can process it.
     bool explicit_flow_control_ = false;
     // Latest intel data retrieved from the StreamInfo.
-    envoy_stream_intel stream_intel_{-1, -1, 0};
+    envoy_stream_intel stream_intel_{-1, -1, 0, 0};
     envoy_final_stream_intel envoy_final_stream_intel_;
     StreamInfo::BytesMeterSharedPtr bytes_meter_;
   };

library/common/jni/jni_utility.cc

@@ -85,12 +85,13 @@ jbyteArray native_data_to_array(JNIEnv* env, envoy_data data) {
 }
 
 jlongArray native_stream_intel_to_array(JNIEnv* env, envoy_stream_intel stream_intel) {
-  jlongArray j_array = env->NewLongArray(3);
+  jlongArray j_array = env->NewLongArray(4);
   jlong* critical_array = static_cast<jlong*>(env->GetPrimitiveArrayCritical(j_array, nullptr));
   RELEASE_ASSERT(critical_array != nullptr, "unable to allocate memory in jni_utility");
   critical_array[0] = static_cast<jlong>(stream_intel.stream_id);
   critical_array[1] = static_cast<jlong>(stream_intel.connection_id);
   critical_array[2] = static_cast<jlong>(stream_intel.attempt_count);
+  critical_array[3] = static_cast<jlong>(stream_intel.received_byte_count);
   // Here '0' (for which there is no named constant) indicates we want to commit the changes back
   // to the JVM and free the c array, where applicable.
   env->ReleasePrimitiveArrayCritical(j_array, critical_array, 0);

library/common/types/c_types.h

@@ -153,6 +153,8 @@ typedef struct {
   int64_t connection_id;
   // The number of internal attempts to carry out a request/operation. 0 if not present.
   uint64_t attempt_count;
+  // The number of bytes received from upstream.
+  uint64_t received_byte_count;
 } envoy_stream_intel;
 
 /**

library/java/io/envoyproxy/envoymobile/engine/EnvoyStreamIntelImpl.java

@@ -6,11 +6,13 @@ class EnvoyStreamIntelImpl implements EnvoyStreamIntel {
   private long streamId;
   private long connectionId;
   private long attemptCount;
+  private long receivedByteCount;
 
   EnvoyStreamIntelImpl(long[] values) {
     streamId = values[0];
     connectionId = values[1];
     attemptCount = values[2];
+    receivedByteCount = values[3];
   }
 
   @Override
@@ -27,4 +29,9 @@ public long getConnectionId() {
   public long getAttemptCount() {
     return attemptCount;
   }
+
+  @Override
+  public long getReceivedByteCount() {
+    return receivedByteCount;
+  }
 }

library/java/io/envoyproxy/envoymobile/engine/types/EnvoyStreamIntel.java

@@ -18,4 +18,8 @@ public interface EnvoyStreamIntel {
    * The number of internal attempts to carry out a request/operation.
    */
   public long getAttemptCount();
+  /*
+   * The number of bytes received from upstream.
+   */
+  public long getReceivedByteCount();
 }

library/java/org/chromium/net/impl/CronetUrlRequest.java

@@ -88,7 +88,6 @@ public final class CronetUrlRequest extends UrlRequestBase {
 
   private final String mUserAgent;
   private final HeadersList mRequestHeaders = new HeadersList();
-  private final List<String> mUrlChain = new ArrayList<>();
   private final CronetUrlRequestContext mRequestContext;
   private final AtomicBoolean mWaitingOnRedirect = new AtomicBoolean(false);
   private final AtomicBoolean mWaitingOnRead = new AtomicBoolean(false);
@@ -131,6 +130,8 @@ public final class CronetUrlRequest extends UrlRequestBase {
 
   /* These change with redirects. */
   private final AtomicReference<EnvoyHTTPStream> mStream = new AtomicReference<>();
+  private final List<String> mUrlChain = new ArrayList<>();
+  private final List<EnvoyFinalStreamIntel> mEnvoyFinalStreamIntels = new ArrayList<>();
   private CronvoyHttpCallbacks mCronvoyCallbacks;
   private String mCurrentUrl;
   private UrlResponseInfoImpl mUrlResponseInfo;
@@ -631,6 +632,25 @@ private boolean streamEnded() {
     return cronvoyCallbacks != null && cronvoyCallbacks.mEndStream;
   }
 
+  private void recordEnvoyFinalStreamIntel(EnvoyFinalStreamIntel envoyFinalStreamIntel) {
+    mEnvoyFinalStreamIntels.add(envoyFinalStreamIntel);
+    long bytesReceived = 0;
+    // This in only called by the network Thread - no concurrency issue.
+    for (EnvoyFinalStreamIntel intel : mEnvoyFinalStreamIntels) {
+      bytesReceived += intel.getReceivedByteCount();
+    }
+    mUrlResponseInfo.setReceivedByteCount(bytesReceived);
+  }
+
+  private void recordEnvoyStreamIntel(EnvoyStreamIntel envoyStreamIntel) {
+    long bytesReceived = envoyStreamIntel.getReceivedByteCount();
+    // This in only called by the network Thread - no concurrency issue.
+    for (EnvoyFinalStreamIntel intel : mEnvoyFinalStreamIntels) {
+      bytesReceived += intel.getReceivedByteCount();
+    }
+    mUrlResponseInfo.setReceivedByteCount(bytesReceived);
+  }
+
   private static class HeadersList extends ArrayList<Map.Entry<String, String>> {}
 
   private static class DirectExecutor implements Executor {
@@ -666,6 +686,7 @@ public Executor getExecutor() {
     @Override
     public void onHeaders(Map<String, List<String>> headers, boolean endStream,
                           EnvoyStreamIntel streamIntel) {
+      recordEnvoyStreamIntel(streamIntel);
       if (isAbandoned()) {
         return;
       }
@@ -731,6 +752,7 @@ public void run() {
 
     @Override
     public void onData(ByteBuffer data, boolean endStream, EnvoyStreamIntel streamIntel) {
+      recordEnvoyStreamIntel(streamIntel);
       if (isAbandoned()) {
         return;
       }
@@ -768,6 +790,7 @@ public void run() {
 
     @Override
     public void onTrailers(Map<String, List<String>> trailers, EnvoyStreamIntel streamIntel) {
+      recordEnvoyStreamIntel(streamIntel);
       if (isAbandoned()) {
         return;
       }
@@ -786,6 +809,7 @@ public void onTrailers(Map<String, List<String>> trailers, EnvoyStreamIntel stre
     @Override
     public void onError(int errorCode, String message, int attemptCount,
                         EnvoyStreamIntel streamIntel, EnvoyFinalStreamIntel finalStreamIntel) {
+      recordEnvoyFinalStreamIntel(finalStreamIntel);
       if (isAbandoned()) {
         return;
       }
@@ -808,6 +832,7 @@ public void onError(int errorCode, String message, int attemptCount,
 
     @Override
     public void onCancel(EnvoyStreamIntel streamIntel, EnvoyFinalStreamIntel finalStreamIntel) {
+      recordEnvoyFinalStreamIntel(finalStreamIntel);
       if (isAbandoned()) {
         return;
       }
@@ -828,6 +853,7 @@ public void onCancel(EnvoyStreamIntel streamIntel, EnvoyFinalStreamIntel finalSt
 
     @Override
     public void onSendWindowAvailable(EnvoyStreamIntel streamIntel) {
+      recordEnvoyStreamIntel(streamIntel);
       if (isAbandoned()) {
         return;
       }
@@ -849,10 +875,10 @@ public void onSendWindowAvailable(EnvoyStreamIntel streamIntel) {
 
     @Override
     public void onComplete(EnvoyStreamIntel streamIntel, EnvoyFinalStreamIntel finalStreamIntel) {
+      recordEnvoyFinalStreamIntel(finalStreamIntel);
       if (isAbandoned()) {
         return;
       }
-      mUrlResponseInfo.setReceivedByteCount(finalStreamIntel.getSentByteCount());
       if (successReady(SucceededState.ON_COMPLETE_RECEIVED)) {
         onSucceeded();
       }
@@ -900,7 +926,7 @@ void readData(int size) {
      */
     void cancel() {
       EnvoyHTTPStream stream = mStream.get();
-      if (isAbandoned() || mEndStream) {
+      if (this != mCronvoyCallbacks || mEndStream) {
         return;
       }
       @CancelState int oldState = mCancelState.getAndSet(CancelState.CANCELLED);
@@ -936,6 +962,7 @@ private void setUrlResponseInfo(Map<String, List<String>> responseHeaders, int r
       // TODO(https://github.com/envoyproxy/envoy-mobile/issues/1426) set receivedByteCount
       // TODO(https://github.com/envoyproxy/envoy-mobile/issues/1622) support proxy
       // TODO(https://github.com/envoyproxy/envoy-mobile/issues/1546) negotiated protocol
+      // TODO(https://github.com/envoyproxy/envoy-mobile/issues/1578) http caching
       mUrlResponseInfo.setResponseValues(
           new ArrayList<>(mUrlChain), responseCode, HttpReason.getReason(responseCode),
           Collections.unmodifiableList(headerList), false, selectedTransport, ":0");

library/kotlin/io/envoyproxy/envoymobile/mocks/MockStream.kt

@@ -15,6 +15,7 @@ class MockStream internal constructor(underlyingStream: MockEnvoyHTTPStream) : S
     override fun getStreamId(): Long { return 0 }
     override fun getConnectionId(): Long { return 0 }
     override fun getAttemptCount(): Long { return 0 }
+    override fun getReceivedByteCount(): Long { return 0 }
   }
 
   private val mockFinalStreamIntel = object : EnvoyFinalStreamIntel {

test/common/integration/client_integration_test.cc

@@ -36,6 +36,7 @@ typedef struct {
   uint32_t on_complete_calls;
   uint32_t on_error_calls;
   uint32_t on_cancel_calls;
+  uint32_t received_byte_count;
   std::string status;
   ConditionalInitializer* terminal_callback;
 } callbacks_called;
@@ -62,12 +63,13 @@ class ClientIntegrationTest : public BaseIntegrationTest,
     });
 
     bridge_callbacks_.context = &cc_;
-    bridge_callbacks_.on_headers = [](envoy_headers c_headers, bool, envoy_stream_intel,
+    bridge_callbacks_.on_headers = [](envoy_headers c_headers, bool, envoy_stream_intel intel,
                                       void* context) -> void* {
       Http::ResponseHeaderMapPtr response_headers = toResponseHeaders(c_headers);
       callbacks_called* cc_ = static_cast<callbacks_called*>(context);
       cc_->on_headers_calls++;
       cc_->status = response_headers->Status()->value().getStringView();
+      cc_->received_byte_count = intel.received_byte_count;
       return nullptr;
     };
     bridge_callbacks_.on_data = [](envoy_data c_data, bool, envoy_stream_intel,
@@ -143,7 +145,7 @@ name: api_listener
   Http::ClientPtr http_client_{};
   envoy_http_callbacks bridge_callbacks_;
   ConditionalInitializer terminal_callback_;
-  callbacks_called cc_ = {0, 0, 0, 0, 0, "", &terminal_callback_};
+  callbacks_called cc_ = {0, 0, 0, 0, 0, 0, "", &terminal_callback_};
 };
 
 INSTANTIATE_TEST_SUITE_P(IpVersions, ClientIntegrationTest,
@@ -206,6 +208,7 @@ TEST_P(ClientIntegrationTest, Basic) {
   ASSERT_EQ(cc_.status, "200");
   ASSERT_EQ(cc_.on_data_calls, 2);
   ASSERT_EQ(cc_.on_complete_calls, 1);
+  ASSERT_EQ(cc_.received_byte_count, 67);
 
   // stream_success gets charged for 2xx status codes.
   test_server_->waitForCounterEq("http.client.stream_success", 1);

test/java/org/chromium/net/CronetUrlRequestTest.java

@@ -219,8 +219,9 @@ public void testRedirectAsync() throws Exception {
     checkResponseInfoHeader(callback.mRedirectResponseInfoList.get(0), "redirect-header",
                             "header-value");
 
+    // Original bytesReceived: 73
     UrlResponseInfo expected = createUrlResponseInfo(
-        new String[] {NativeTestServer.getRedirectURL()}, "Found", 302, 73, "Content-Length", "92",
+        new String[] {NativeTestServer.getRedirectURL()}, "Found", 302, -1, "Content-Length", "92",
         "Location", "/success.txt", "redirect-header", "header-value");
     mTestRule.assertResponseEquals(expected, callback.mRedirectResponseInfoList.get(0));
 
@@ -266,9 +267,10 @@ public void testRedirectAsync() throws Exception {
     assertEquals(ResponseStep.ON_SUCCEEDED, callback.mResponseStep);
     assertEquals(NativeTestServer.SUCCESS_BODY, callback.mResponseAsString);
 
+    // Original bytesReceived: 258
     UrlResponseInfo urlResponseInfo = createUrlResponseInfo(
         new String[] {NativeTestServer.getRedirectURL(), NativeTestServer.getSuccessURL()}, "OK",
-        200, 258, "Content-Length", "20", "Content-Type", "text/plain",
+        200, -1, "Content-Length", "20", "Content-Type", "text/plain",
         "Access-Control-Allow-Origin", "*", "header-name", "header-value", "multi-header-name",
         "header-value1", "multi-header-name", "header-value2");
 
@@ -666,17 +668,19 @@ public void testMockMultiRedirect() throws Exception {
     assertEquals(2, callback.mRedirectResponseInfoList.size());
 
     // Check first redirect (multiredirect.html -> redirect.html)
+    // Original receivedBytes: 76
     UrlResponseInfo firstExpectedResponseInfo = createUrlResponseInfo(
-        new String[] {NativeTestServer.getMultiRedirectURL()}, "Found", 302, 76, "Content-Length",
+        new String[] {NativeTestServer.getMultiRedirectURL()}, "Found", 302, -1, "Content-Length",
         "92", "Location", "/redirect.html", "redirect-header0", "header-value");
     UrlResponseInfo firstRedirectResponseInfo = callback.mRedirectResponseInfoList.get(0);
     mTestRule.assertResponseEquals(firstExpectedResponseInfo, firstRedirectResponseInfo);
 
     // Check second redirect (redirect.html -> success.txt)
+    // Original receivedBytes: 334
     UrlResponseInfo secondExpectedResponseInfo = createUrlResponseInfo(
         new String[] {NativeTestServer.getMultiRedirectURL(), NativeTestServer.getRedirectURL(),
                       NativeTestServer.getSuccessURL()},
-        "OK", 200, 334, "Content-Length", "20", "Content-Type", "text/plain",
+        "OK", 200, -1, "Content-Length", "20", "Content-Type", "text/plain",
         "Access-Control-Allow-Origin", "*", "header-name", "header-value", "multi-header-name",
         "header-value1", "multi-header-name", "header-value2");
 
@@ -693,7 +697,7 @@ public void testMockNotFound() throws Exception {
     TestUrlRequestCallback callback = startAndWaitForComplete(NativeTestServer.getNotFoundURL());
     UrlResponseInfo expected =
         createUrlResponseInfo(new String[] {NativeTestServer.getNotFoundURL()}, "Not Found", 404,
-                              140, "Content-Length", "96");
+                              142, "Content-Length", "96");
     mTestRule.assertResponseEquals(expected, callback.mResponseInfo);
     assertTrue(callback.mHttpResponseDataLength != 0);
     assertEquals(0, callback.mRedirectCount);

test/java/org/chromium/net/testing/CronetTestRule.java

@@ -68,7 +68,7 @@ public static class CronetTestFramework {
 
     private static ExperimentalCronetEngine createEngine(Context context) {
       ExperimentalCronetEngine.Builder builder = new ExperimentalCronetEngine.Builder(context);
-      ((CronetEngineBuilderImpl)builder.getBuilderDelegate()).setLogLevel("info");
+      ((CronetEngineBuilderImpl)builder.getBuilderDelegate()).setLogLevel("off");
       return builder.enableQuic(true).build();
     }
 
@@ -253,8 +253,10 @@ public void assertResponseEquals(UrlResponseInfo expected, UrlResponseInfo actua
     assertEquals(expected.getUrl(), actual.getUrl());
     // Transferred bytes and proxy server are not supported in pure java
     if (!testingJavaImpl()) {
-      // TODO("https://github.com/envoyproxy/envoy-mobile/issues/1426"): uncomment the assert
-      // assertEquals(expected.getReceivedByteCount(), actual.getReceivedByteCount());
+      // TODO("https://github.com/envoyproxy/envoy-mobile/issues/1426"): remove the "if" crutch
+      if (expected.getReceivedByteCount() >= 0) {
+        assertEquals(expected.getReceivedByteCount(), actual.getReceivedByteCount());
+      }
       assertEquals(expected.getProxyServer(), actual.getProxyServer());
       // This is a place where behavior intentionally differs between native and java
       assertEquals(expected.getNegotiatedProtocol(), actual.getNegotiatedProtocol());