[Feat]新增动态切换QUIC

foudations/mogo-network/build.gradle

@@ -7,6 +7,7 @@ apply from: new File(rootProject.rootDir, "gradle/upload.gradle").toString()
 
 android {
     compileSdkVersion rootProject.ext.android.compileSdkVersion
+    ndkVersion '21.4.7075529'
 
     defaultConfig {
         minSdkVersion rootProject.ext.android.minSdkVersion
@@ -16,6 +17,10 @@ android {
         versionName "1.0"
 
         consumerProguardFiles "consumer-rules.pro"
+
+        ndk {
+            abiFilters "arm64-v8a"
+        }
     }
 
     buildTypes {
@@ -56,6 +61,10 @@ dependencies {
     api rootProject.ext.dependencies.retrofitconvertergson
     api rootProject.ext.dependencies.retrofitconverterscalars
     compileOnly 'com.elegant.network:network:1.1.28'
+    implementation files('libs/cronet_api.jar')
+    implementation files('libs/cronet_impl_common_java.jar')
+    implementation files('libs/cronet_impl_native_java.jar')
+    implementation 'com.google.guava:guava:29.0-android'
 
     if (Boolean.valueOf(RELEASE)) {
         api "com.mogo.cloud:passport:${MOGO_PASSPORT_VERSION}"

foudations/mogo-network/src/main/java/com/mogo/cloud/network/OkHttpFactory.kt

@@ -5,6 +5,7 @@ import com.mogo.cloud.network.NetConstants.Companion.READ_TIMEOUT
 import com.mogo.cloud.network.NetConstants.Companion.WRITE_TIMEOUT
 import com.mogo.cloud.network.SSLSocketFactoryUtils.createSSLSocketFactory
 import com.mogo.cloud.network.SSLSocketFactoryUtils.createTrustAllManager
+import com.mogo.cloud.network.cronet.CronetInterceptor
 import com.mogo.cloud.network.interceptor.HttpHeaderInterceptor
 import com.mogo.cloud.network.interceptor.HttpLoggingInterceptor
 import com.mogo.cloud.network.interceptor.HttpPassportInterceptor
@@ -23,9 +24,10 @@ class OkHttpFactory private constructor() {
 
         val okHttpClient: OkHttpClient by lazy {
             OkHttpClient.Builder()
-                    .addNetworkInterceptor(HttpHeaderInterceptor())
-                    .addNetworkInterceptor(HttpLoggingInterceptor())
+                    .addInterceptor(HttpHeaderInterceptor())
                     .addInterceptor(HttpPassportInterceptor())
+                    .addInterceptor(CronetInterceptor())
+                    .addInterceptor(HttpLoggingInterceptor())
                     .sslSocketFactory(createSSLSocketFactory(), createTrustAllManager())
                     .hostnameVerifier(SSLSocketFactoryUtils.TrustAllHostnameVerifier())
                     .connectTimeout(CONNECT_TIMEOUT, TimeUnit.MILLISECONDS)

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetCallFactory.java

@@ -0,0 +1,328 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import android.util.Log;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import java.io.IOException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import okhttp3.Call;
+import okhttp3.Callback;
+import okhttp3.Request;
+import okhttp3.Response;
+import okio.AsyncTimeout;
+import okio.Timeout;
+import org.chromium.net.CronetEngine;
+
+/** A {@link Call.Factory} implementation using Cronet as the transport layer. */
+public final class CronetCallFactory implements Call.Factory {
+
+  private static final String TAG = "CronetCallFactory";
+
+  private final RequestResponseConverter converter;
+  private final ExecutorService responseCallbackExecutor;
+  private final int readTimeoutMillis;
+  private final int writeTimeoutMillis;
+  private final int callTimeoutMillis;
+
+  private CronetCallFactory(
+      RequestResponseConverter converter,
+      ExecutorService responseCallbackExecutor,
+      int readTimeoutMillis,
+      int writeTimeoutMillis,
+      int callTimeoutMillis) {
+    checkArgument(readTimeoutMillis >= 0, "Read timeout mustn't be negative!");
+    checkArgument(writeTimeoutMillis >= 0, "Write timeout mustn't be negative!");
+    checkArgument(callTimeoutMillis >= 0, "Call timeout mustn't be negative!");
+
+    this.converter = converter;
+    this.responseCallbackExecutor = responseCallbackExecutor;
+    this.readTimeoutMillis = readTimeoutMillis;
+    this.writeTimeoutMillis = writeTimeoutMillis;
+    this.callTimeoutMillis = callTimeoutMillis;
+  }
+
+  public static Builder newBuilder(CronetEngine cronetEngine) {
+    return new Builder(cronetEngine);
+  }
+
+  @Override
+  public Call newCall(Request request) {
+    return new CronetCall(request, this, converter, responseCallbackExecutor);
+  }
+
+  private static class CronetCall implements Call {
+
+    private final Request okHttpRequest;
+    private final CronetCallFactory motherFactory;
+    private final RequestResponseConverter converter;
+    private final ExecutorService responseCallbackExecutor;
+
+    private final AtomicBoolean executed = new AtomicBoolean();
+    private final AtomicBoolean canceled = new AtomicBoolean();
+    private final AtomicReference<RequestResponseConverter.CronetRequestAndOkHttpResponse> convertedRequestAndResponse =
+        new AtomicReference<>();
+    private final AsyncTimeout timeout;
+
+    private CronetCall(
+        Request okHttpRequest,
+        CronetCallFactory motherFactory,
+        RequestResponseConverter converter,
+        ExecutorService responseCallbackExecutor) {
+      this.okHttpRequest = okHttpRequest;
+      this.motherFactory = motherFactory;
+      this.converter = converter;
+      this.responseCallbackExecutor = responseCallbackExecutor;
+
+      this.timeout =
+          new AsyncTimeout() {
+            @Override
+            protected void timedOut() {
+              cancel();
+            }
+          };
+      timeout.timeout(motherFactory.callTimeoutMillis, MILLISECONDS);
+    }
+
+    @Override
+    public Request request() {
+      return okHttpRequest;
+    }
+
+    @Override
+    public Response execute() throws IOException {
+      evaluateExecutionPreconditions();
+      try {
+        timeout.enter();
+        RequestResponseConverter.CronetRequestAndOkHttpResponse requestAndOkHttpResponse =
+            converter.convert(
+                request(), motherFactory.readTimeoutMillis, motherFactory.writeTimeoutMillis);
+        convertedRequestAndResponse.set(requestAndOkHttpResponse);
+
+        startRequestIfNotCanceled();
+
+        return toCronetCallFactoryResponse(this, requestAndOkHttpResponse.getResponse());
+      } catch (RuntimeException | IOException e) {
+        // If the request finished successfully don't exit the timeout yet. Reading the body also
+        // needs to be considered and the body object will take care of exiting it. See
+        // toCronetCallFactoryResponse() for details.
+        timeout.exit();
+        throw e;
+      }
+    }
+
+    @Override
+    public void enqueue(Callback responseCallback) {
+      try {
+        timeout.enter();
+        evaluateExecutionPreconditions();
+        RequestResponseConverter.CronetRequestAndOkHttpResponse requestAndOkHttpResponse =
+            converter.convert(
+                request(), motherFactory.readTimeoutMillis, motherFactory.writeTimeoutMillis);
+        convertedRequestAndResponse.set(requestAndOkHttpResponse);
+        CronetCall call = this;
+
+        Futures.addCallback(
+            requestAndOkHttpResponse.getResponseAsync(),
+            new FutureCallback<Response>() {
+              @Override
+              public void onSuccess(Response result) {
+                try {
+                  responseCallback.onResponse(call, toCronetCallFactoryResponse(call, result));
+                } catch (IOException e) {
+                  // The call factory doesn't really mind this - the application code
+                  // threw an exception while handling the response, they should have taken care
+                  // of it. Just logging the error is consistent with plain OkHttp implementation.
+                  Log.i(TAG, "Callback failure for " + toLoggableString(), e);
+                }
+              }
+
+              @Override
+              public void onFailure(Throwable t) {
+                if (t instanceof IOException) {
+                  responseCallback.onFailure(call, (IOException) t);
+                } else {
+                  responseCallback.onFailure(call, new IOException(t));
+                }
+              }
+            },
+            responseCallbackExecutor);
+
+        startRequestIfNotCanceled();
+      } catch (IOException e) {
+        // If the request finished successfully don't exit the timeout yet. Reading the body also
+        // needs to be considered and the body object will take care of exiting it. See
+        // toCronetCallFactoryResponse() for details.
+        timeout.exit();
+        responseCallback.onFailure(this, e);
+      }
+    }
+
+    @Override
+    public Call clone() {
+      return motherFactory.newCall(request());
+    }
+
+    @Override
+    public void cancel() {
+      if (canceled.getAndSet(true)) {
+        // already canceled
+        return;
+      }
+      RequestResponseConverter.CronetRequestAndOkHttpResponse localConverted = convertedRequestAndResponse.get();
+      if (localConverted != null) {
+        localConverted.getRequest().cancel();
+      } // else the cancel signal will be picked up by the execute() / enqueue() methods.
+    }
+
+    @Override
+    public boolean isExecuted() {
+      return executed.get();
+    }
+
+    @Override
+    public boolean isCanceled() {
+      return canceled.get();
+    }
+
+    @Override
+    public Timeout timeout() {
+      return timeout;
+    }
+
+    private String toLoggableString() {
+      return "call to " + request().url().redact();
+    }
+
+    /**
+     * Verifies that the call can be executed and sets the state of the call to "being executed".
+     *
+     * @throws IllegalStateException if the request has already been executed.
+     * @throws IOException if the request was canceled
+     */
+    private void evaluateExecutionPreconditions() throws IOException {
+      if (canceled.get()) {
+        throw new IOException("Can't execute canceled requests");
+      }
+      checkState(!executed.getAndSet(true), "Already Executed");
+    }
+
+    private void startRequestIfNotCanceled() {
+      RequestResponseConverter.CronetRequestAndOkHttpResponse requestAndOkHttpResponse = convertedRequestAndResponse.get();
+      checkState(requestAndOkHttpResponse != null, "convertedRequestAndResponse must be set!");
+
+      // There might be a race between the execution and cancellation
+      // evaluateExecutionPreconditions check didn't capture and cancel() might have missed that
+      // as well. Check once again that the request isn't canceled.
+      // This way, no matter how the instructions of the two threads are interleaved, we always end
+      // up with a serialized-like outcome (either cancel() was fully run before execute(), or vice
+      // versa).
+
+      // Thread 1 (cancel() call)           | Thread 2 (execute() call)
+      // -------------------------------------------------------------------------------
+      // canceled = true                    | if (canceled) throw;
+      // convertedRequest?.cancel()         | convertedRequest = convert(request)
+      //                                    | if (canceled) convertedRequest.cancel()
+      if (canceled.get()) {
+        requestAndOkHttpResponse.getRequest().cancel();
+      } else {
+        requestAndOkHttpResponse.getRequest().start();
+      }
+    }
+  }
+
+  private static Response toCronetCallFactoryResponse(CronetCall call, Response response) {
+    checkNotNull(response.body());
+
+    return response
+        .newBuilder()
+        .body(
+            new CronetTransportResponseBody(response.body()) {
+              @Override
+              void customCloseHook() {
+                call.timeout.exit();
+              }
+            })
+        .build();
+  }
+
+  public static final class Builder
+      extends RequestResponseConverterBasedBuilder<Builder, CronetCallFactory> {
+    private static final int DEFAULT_READ_WRITE_TIMEOUT_MILLIS = 10000;
+
+    private int readTimeoutMillis = DEFAULT_READ_WRITE_TIMEOUT_MILLIS;
+    private int writeTimeoutMillis = DEFAULT_READ_WRITE_TIMEOUT_MILLIS;
+    private int callTimeoutMillis = 0; // No timeout
+    private ExecutorService callbackExecutorService = null;
+
+    Builder(CronetEngine cronetEngine) {
+      super(cronetEngine, Builder.class);
+    }
+
+    public Builder setReadTimeoutMillis(int readTimeoutMillis) {
+      checkArgument(readTimeoutMillis >= 0, "Read timeout mustn't be negative!");
+      this.readTimeoutMillis = readTimeoutMillis;
+      return this;
+    }
+
+    public Builder setWriteTimeoutMillis(int writeTimeoutMillis) {
+      checkArgument(writeTimeoutMillis >= 0, "Write timeout mustn't be negative!");
+      this.writeTimeoutMillis = writeTimeoutMillis;
+      return this;
+    }
+
+    public Builder setCallbackExecutorService(ExecutorService callbackExecutorService) {
+      checkNotNull(callbackExecutorService);
+      this.callbackExecutorService = callbackExecutorService;
+      return this;
+    }
+
+    public Builder setCallTimeoutMillis(int callTimeoutMillis) {
+      checkArgument(callTimeoutMillis >= 0, "Call timeout mustn't be negative!");
+      this.callTimeoutMillis = callTimeoutMillis;
+
+      return this;
+    }
+
+    @Override
+    CronetCallFactory build(RequestResponseConverter converter) {
+      ExecutorService localCallbackExecutorService;
+      if (callbackExecutorService == null) {
+        // Consistent with OkHttp impl
+        localCallbackExecutorService = Executors.newCachedThreadPool();
+      } else {
+        localCallbackExecutorService = callbackExecutorService;
+      }
+
+      return new CronetCallFactory(
+          converter,
+          localCallbackExecutorService,
+          readTimeoutMillis,
+          writeTimeoutMillis,
+          callTimeoutMillis);
+    }
+  }
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetInterceptor.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import android.util.Log;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+import okhttp3.Call;
+import okhttp3.Interceptor;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+
+import org.chromium.net.UrlRequest;
+
+/**
+ * An OkHttp interceptor that redirects HTTP traffic to use Cronet instead of using the OkHttp
+ * network stack.
+ *
+ * <p>The interceptor should be used as the last application interceptor to ensure that all other
+ * interceptors are visited before sending the request on wire and after a response is returned.
+ *
+ * <p>The interceptor is a plug-and-play replacement for the OkHttp stack for the most part,
+ * however, there are some caveats to keep in mind:
+ *
+ * <ol>
+ *   <li>The entirety of OkHttp core is bypassed. This includes caching configuration and network
+ *       interceptors.
+ *   <li>Some response fields are not being populated due to mismatches between Cronet's and
+ *       OkHttp's architecture. TODO(danstahr): add a concrete list).
+ * </ol>
+ */
+public final class CronetInterceptor implements Interceptor, AutoCloseable {
+    private static final String TAG = "CronetInterceptor";
+
+    private static final int CANCELLATION_CHECK_INTERVAL_MILLIS = 500;
+    private final Map<Call, UrlRequest> activeCalls = new ConcurrentHashMap<>();
+    private final ScheduledExecutorService scheduledExecutor = new ScheduledThreadPoolExecutor(1);
+    private ScheduledFuture<?> unusedFuture;
+
+    public CronetInterceptor() {
+        // TODO(danstahr): There's no other way to know if the call is canceled but polling
+        //  (https://github.com/square/okhttp/issues/7164).
+        unusedFuture = scheduledExecutor.scheduleAtFixedRate(
+                () -> {
+                    Iterator<Entry<Call, UrlRequest>> activeCallsIterator =
+                            activeCalls.entrySet().iterator();
+
+                    while (activeCallsIterator.hasNext()) {
+                        try {
+                            Entry<Call, UrlRequest> activeCall = activeCallsIterator.next();
+                            if (activeCall.getKey().isCanceled()) {
+                                activeCallsIterator.remove();
+                                activeCall.getValue().cancel();
+                            }
+                        } catch (RuntimeException e) {
+                            Log.w(TAG, "Unable to propagate cancellation status", e);
+                        }
+                    }
+                },
+                CANCELLATION_CHECK_INTERVAL_MILLIS,
+                CANCELLATION_CHECK_INTERVAL_MILLIS,
+                MILLISECONDS);
+    }
+
+    @Override
+    public Response intercept(Chain chain) throws IOException {
+        if (!QuicConfig.INSTANCE.getEnable() || QuicConfig.INSTANCE.getEngine() == null || QuicConfig.INSTANCE.getConverter() == null) {
+            return chain.proceed(chain.request());
+        }
+        if (chain.call().isCanceled()) {
+            throw new IOException("Canceled");
+        }
+        Request request = chain.request();
+        RequestResponseConverter.CronetRequestAndOkHttpResponse requestAndOkHttpResponse = QuicConfig.INSTANCE.getConverter().convert(
+                request, chain.readTimeoutMillis(), chain.writeTimeoutMillis());
+
+        activeCalls.put(chain.call(), requestAndOkHttpResponse.getRequest());
+
+        try {
+            requestAndOkHttpResponse.getRequest().start();
+            return toInterceptorResponse(requestAndOkHttpResponse.getResponse(), chain.call());
+        } catch (RuntimeException | IOException e) {
+            // If the response is retrieved successfully the caller is responsible for closing
+            // the response, which will remove it from the active calls map.
+            activeCalls.remove(chain.call());
+            throw e;
+        }
+    }
+
+    @Override
+    public void close() {
+        scheduledExecutor.shutdown();
+    }
+
+    private Response toInterceptorResponse(Response response, Call call) {
+        checkNotNull(response.body());
+
+        if (response.body() instanceof CronetInterceptorResponseBody) {
+            return response;
+        }
+
+        return response
+                .newBuilder()
+                .body(new CronetInterceptorResponseBody(response.body(), call))
+                .build();
+    }
+
+    private class CronetInterceptorResponseBody extends CronetTransportResponseBody {
+        private final Call call;
+
+        private CronetInterceptorResponseBody(ResponseBody delegate, Call call) {
+            super(delegate);
+            this.call = call;
+        }
+
+        @Override
+        void customCloseHook() {
+            activeCalls.remove(call);
+        }
+    }
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetTimeoutException.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import java.io.IOException;
+
+public class CronetTimeoutException extends IOException {}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetTransportResponseBody.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import androidx.annotation.Nullable;
+import okhttp3.MediaType;
+import okhttp3.ResponseBody;
+import okio.BufferedSource;
+
+abstract class CronetTransportResponseBody extends ResponseBody {
+
+  private final ResponseBody delegate;
+
+  protected CronetTransportResponseBody(ResponseBody delegate) {
+    this.delegate = delegate;
+  }
+
+  @Nullable
+  @Override
+  public final MediaType contentType() {
+    return delegate.contentType();
+  }
+
+  @Override
+  public final long contentLength() {
+    return delegate.contentLength();
+  }
+
+  @Override
+  public final BufferedSource source() {
+    return delegate.source();
+  }
+
+  @Override
+  public final void close() {
+    delegate.close();
+    customCloseHook();
+  }
+
+  abstract void customCloseHook();
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/OkHttpBridgeRequestCallback.java

@@ -0,0 +1,313 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import androidx.annotation.Nullable;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import java.io.IOException;
+import java.net.ProtocolException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import okio.Buffer;
+import okio.Source;
+import okio.Timeout;
+import org.chromium.net.CronetException;
+import org.chromium.net.UrlRequest;
+import org.chromium.net.UrlResponseInfo;
+
+/**
+ * An implementation of Cronet's callback. This is the heart of the bridge and deals with most of
+ * the async-sync paradigm translation.
+ *
+ * <p>Translating the UrlResponseInfo is relatively straightforward as the entire object is
+ * available immediately and is relatively small, so it can easily fit in memory.
+ *
+ * <p>Translating the body is a bit more tricky because of the mismatch between OkHttp and Cronet
+ * designs. We invoke Cronet's read and wait for the result using synchronization primitives (see
+ * BodySource implementation). The implementation is assuming that there's always at most one read()
+ * request in flight (which is safe to assume), and relies on reasonable fairness of thread
+ * scheduling, especially when handling cancellations.
+ */
+class OkHttpBridgeRequestCallback extends UrlRequest.Callback {
+
+  /**
+   * The byte buffer capacity for reading Cronet response bodies. Each response callback will
+   * allocate its own buffer of this size once the response starts being processed.
+   */
+  private static final int CRONET_BYTE_BUFFER_CAPACITY = 32 * 1024;
+
+  /** A bridge between Cronet's asynchronous callbacks and OkHttp's blocking stream-like reads. */
+  private final SettableFuture<Source> bodySourceFuture = SettableFuture.create();
+
+  /** Signal whether the request is finished and the response has been fully read. */
+  private final AtomicBoolean finished = new AtomicBoolean(false);
+
+  /** Signal whether the request was canceled. */
+  private final AtomicBoolean canceled = new AtomicBoolean(false);
+
+  /**
+   * An internal, blocking, thread safe way of passing data between the callback methods and {@link
+   * #bodySourceFuture}.
+   *
+   * <p>Has a capacity of 2 - at most one slot for a read result and at most 1 slot for cancellation
+   * signal, this guarantees that all inserts are non blocking.
+   */
+  private final BlockingQueue<CallbackResult> callbackResults = new ArrayBlockingQueue<>(2);
+
+  /** The response headers. */
+  private final SettableFuture<UrlResponseInfo> headersFuture = SettableFuture.create();
+
+  /** The read timeout as specified by OkHttp. * */
+  private final long readTimeoutMillis;
+
+  /** The previous responses as reported to {@link #onRedirectReceived}, from oldest to newest. * */
+  private final List<UrlResponseInfo> urlResponseInfoChain = new ArrayList<>();
+
+  private final RedirectStrategy redirectStrategy;
+
+  /** The request being processed. Set when the request is first seen by the callback. */
+  private volatile UrlRequest request;
+
+  OkHttpBridgeRequestCallback(long readTimeoutMillis, RedirectStrategy redirectStrategy) {
+    checkArgument(readTimeoutMillis >= 0);
+
+    // So that we don't have to special case infinity. Int.MAX_VALUE is ~infinity for all practical
+    // use cases.
+    if (readTimeoutMillis == 0) {
+      this.readTimeoutMillis = Integer.MAX_VALUE;
+    } else {
+      this.readTimeoutMillis = readTimeoutMillis;
+    }
+    this.redirectStrategy = redirectStrategy;
+  }
+
+  /** Returns the {@link UrlResponseInfo} for the request associated with this callback. */
+  ListenableFuture<UrlResponseInfo> getUrlResponseInfo() {
+    return headersFuture;
+  }
+
+  /**
+   * Returns the OkHttp {@link Source} for the request associated with this callback.
+   *
+   * <p>Note that retrieving data from the {@code Source} instance might block further as the
+   * response body is streamed.
+   */
+  ListenableFuture<Source> getBodySource() {
+    return bodySourceFuture;
+  }
+
+  List<UrlResponseInfo> getUrlResponseInfoChain() {
+    return Collections.unmodifiableList(urlResponseInfoChain);
+  }
+
+  @Override
+  public void onRedirectReceived(
+      UrlRequest urlRequest, UrlResponseInfo urlResponseInfo, String nextUrl) {
+    // We shouldn't follow redirects - pass the given UrlResponseInfo as the ultimate result
+    if (!redirectStrategy.followRedirects()) {
+      checkState(headersFuture.set(urlResponseInfo));
+      // Note: This might not match the content length headers but we have no way of accessing
+      // the actual body with current Cronet's APIs (see RedirectStrategy).
+      checkState(bodySourceFuture.set(new Buffer()));
+      urlRequest.cancel();
+      return;
+    }
+
+    // We should follow redirects and we haven't hit the cap yet
+    urlResponseInfoChain.add(urlResponseInfo);
+    if (urlResponseInfo.getUrlChain().size() <= redirectStrategy.numberOfRedirectsToFollow()) {
+      urlRequest.followRedirect();
+      return;
+    }
+
+    // Cap reached - cancel the request and fail. Exception crafted to match OkHttp.
+    urlRequest.cancel();
+
+    IOException e =
+        new ProtocolException(
+            "Too many follow-up requests: " + (redirectStrategy.numberOfRedirectsToFollow() + 1));
+    headersFuture.setException(e);
+    bodySourceFuture.setException(e);
+  }
+
+  @Override
+  public void onResponseStarted(UrlRequest urlRequest, UrlResponseInfo urlResponseInfo) {
+    request = urlRequest;
+
+    checkState(headersFuture.set(urlResponseInfo));
+    checkState(bodySourceFuture.set(new CronetBodySource()));
+  }
+
+  @Override
+  public void onReadCompleted(
+      UrlRequest urlRequest, UrlResponseInfo urlResponseInfo, ByteBuffer byteBuffer) {
+    callbackResults.add(new CallbackResult(CallbackStep.ON_READ_COMPLETED, byteBuffer, null));
+  }
+
+  @Override
+  public void onSucceeded(UrlRequest urlRequest, UrlResponseInfo urlResponseInfo) {
+    callbackResults.add(new CallbackResult(CallbackStep.ON_SUCCESS, null, null));
+  }
+
+  @Override
+  public void onFailed(UrlRequest urlRequest, UrlResponseInfo urlResponseInfo, CronetException e) {
+    // If this was called before we start reading the body, the exception will
+    // propagate in the future providing headers and the body wrapper.
+    if (headersFuture.setException(e) && bodySourceFuture.setException(e)) {
+      return;
+    }
+
+    // If this was called as a reaction to a read() call, the read result will propagate
+    // the exception.
+    callbackResults.add(new CallbackResult(CallbackStep.ON_FAILED, null, e));
+  }
+
+  @Override
+  public void onCanceled(UrlRequest urlRequest, UrlResponseInfo responseInfo) {
+    canceled.set(true);
+    callbackResults.add(new CallbackResult(CallbackStep.ON_CANCELED, null, null));
+
+    // If there's nobody listening it's possible that the cancellation happened before we even
+    // received anything from the server. In that case inform the thread that's awaiting server
+    // response about the cancellation as well. This becomes a no-op if the futures
+    // were already set.
+    IOException e = new IOException("The request was canceled!");
+    headersFuture.setException(e);
+    bodySourceFuture.setException(e);
+  }
+
+  private class CronetBodySource implements Source {
+
+    private ByteBuffer buffer = ByteBuffer.allocateDirect(CRONET_BYTE_BUFFER_CAPACITY);
+
+    /** Whether the close() method has been called. */
+    private volatile boolean closed = false;
+
+    @Override
+    public long read(Buffer sink, long byteCount) throws IOException {
+      if (canceled.get()) {
+        throw new IOException("The request was canceled!");
+      }
+
+      // Using IAE instead of NPE (checkNotNull) for okio.RealBufferedSource consistency
+      checkArgument(sink != null, "sink == null");
+      checkArgument(byteCount >= 0, "byteCount < 0: %s", byteCount);
+      checkState(!closed, "closed");
+
+      if (finished.get()) {
+        return -1;
+      }
+
+      if (byteCount < buffer.limit()) {
+        buffer.limit((int) byteCount);
+      }
+
+      request.read(buffer);
+
+      CallbackResult result;
+      try {
+        result = callbackResults.poll(readTimeoutMillis, MILLISECONDS);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        result = null;
+      }
+
+      if (result == null) {
+        // Either readResult.poll() was interrupted or it timed out.
+        request.cancel();
+        throw new CronetTimeoutException();
+      }
+
+      switch (result.callbackStep) {
+        // We null the buffer in final statuses to allow fast GC of the buffer even if the callback
+        // is still in use.
+        case ON_FAILED:
+          finished.set(true);
+          buffer = null;
+          throw new IOException(result.exception);
+        case ON_SUCCESS:
+          finished.set(true);
+          buffer = null;
+          return -1;
+        case ON_CANCELED:
+          // The canceled flag is already set by the onCanceled method
+          // so not setting it here.
+
+          buffer = null;
+          throw new IOException("The request was canceled!");
+        case ON_READ_COMPLETED:
+          result.buffer.flip();
+          int bytesWritten = sink.write(result.buffer);
+          result.buffer.clear();
+          return bytesWritten;
+      }
+
+      throw new AssertionError("The switch block above is exhaustive!");
+    }
+
+    @Override
+    public Timeout timeout() {
+      // TODO(danstahr): This should likely respect the OkHttp timeout somehow
+      return Timeout.NONE;
+    }
+
+    @Override
+    public void close() {
+      if (closed) {
+        return;
+      }
+      closed = true;
+      if (!finished.get()) {
+        request.cancel();
+      }
+    }
+  }
+
+  private static class CallbackResult {
+    private final CallbackStep callbackStep;
+    @Nullable
+    private final ByteBuffer buffer;
+    @Nullable private final CronetException exception;
+
+    private CallbackResult(
+        CallbackStep callbackStep,
+        @Nullable ByteBuffer buffer,
+        @Nullable CronetException exception) {
+      this.callbackStep = callbackStep;
+      this.buffer = buffer;
+      this.exception = exception;
+    }
+  }
+
+  private enum CallbackStep {
+    ON_READ_COMPLETED,
+    ON_SUCCESS,
+    ON_FAILED,
+    ON_CANCELED
+  }
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/QuicConfig.kt

@@ -0,0 +1,59 @@
+package com.mogo.cloud.network.cronet
+
+import android.annotation.SuppressLint
+import android.content.Context
+import org.chromium.net.CronetEngine
+import java.util.concurrent.Executors
+
+@SuppressLint("StaticFieldLeak")
+object QuicConfig {
+    private var engine: CronetEngine? = null
+    private var converter: RequestResponseConverter? = null
+
+    @Volatile
+    private var enable: Boolean = false
+
+    fun init(context: Context) {
+        if (engine == null) {
+            engine = CronetEngine.Builder(context.applicationContext)
+                .enableQuic(true)
+                .enableHttp2(true)
+                .setUserAgent("OkHttp With Cronet")
+                .build()
+        }
+        if (converter == null) {
+            converter = RequestResponseConverter(
+                engine,
+                Executors.newFixedThreadPool(4),  // There must always be enough executors to blocking-read the OkHttp request bodies
+                // otherwise deadlocks can occur.
+                RequestBodyConverterImpl.create(Executors.newCachedThreadPool()),
+                ResponseConverter(),
+                RedirectStrategy.defaultStrategy()
+            )
+        }
+    }
+
+    fun getEnable(): Boolean {
+        return enable
+    }
+
+    fun setEnable(context: Context, enable: Boolean) {
+        when {
+            enable -> {
+                init(context)
+                this.enable = true
+            }
+            else -> {
+                this.enable = false
+            }
+        }
+    }
+
+    fun getEngine(): CronetEngine? {
+        return engine
+    }
+
+    fun getConverter(): RequestResponseConverter? {
+        return converter
+    }
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RedirectStrategy.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+/** Defines a redirect strategy for the Cronet OkHttp transport layer. */
+public abstract class RedirectStrategy {
+
+  /** The default number of redirects to follow. Should be less than the Chromium wide safeguard. */
+  private static final int DEFAULT_REDIRECTS = 16;
+
+  /**
+   * Returns whether redirects should be followed at all. If set to false, the redirect response
+   * will be returned.
+   */
+  abstract boolean followRedirects();
+
+  /**
+   * Returns the maximum number of redirects to follow. If more redirects are attempted an exception
+   * should be thrown by the component handling the request. Shouldn't be called at all if {@link
+   * #followRedirects()} return false.
+   */
+  abstract int numberOfRedirectsToFollow();
+
+  /**
+   * Returns a strategy which will not follow redirects.
+   *
+   * <p>Note that because of Cronet's limitations
+   * (https://developer.android.com/guide/topics/connectivity/cronet/lifecycle#overview) it is
+   * impossible to retrieve the body of a redirect response. As a result, a dummy empty body will
+   * always be provided.
+   */
+  public static RedirectStrategy withoutRedirects() {
+    return WithoutRedirectsHolder.INSTANCE;
+  }
+
+  /**
+   * Returns a strategy which will follow redirects up to {@link #DEFAULT_REDIRECTS} times. If more
+   * redirects are attempted an exception is thrown.
+   */
+  public static RedirectStrategy defaultStrategy() {
+    return DefaultRedirectsHolder.INSTANCE;
+  }
+
+  private static class WithoutRedirectsHolder {
+    private static final RedirectStrategy INSTANCE =
+        new RedirectStrategy() {
+          @Override
+          boolean followRedirects() {
+            return false;
+          }
+
+          @Override
+          int numberOfRedirectsToFollow() {
+            throw new UnsupportedOperationException();
+          }
+        };
+  }
+
+  private static class DefaultRedirectsHolder {
+    private static final RedirectStrategy INSTANCE =
+        new RedirectStrategy() {
+          @Override
+          boolean followRedirects() {
+            return true;
+          }
+
+          @Override
+          int numberOfRedirectsToFollow() {
+            return DEFAULT_REDIRECTS;
+          }
+        };
+  }
+
+  private RedirectStrategy() {}
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestBodyConverter.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import java.io.IOException;
+import okhttp3.RequestBody;
+import org.chromium.net.UploadDataProvider;
+
+/** An interface for classes converting from OkHttp to Cronet request bodies. */
+interface RequestBodyConverter {
+  UploadDataProvider convertRequestBody(RequestBody requestBody, int writeTimeoutMillis)
+      throws IOException;
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestBodyConverterImpl.java

@@ -0,0 +1,335 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import androidx.annotation.VisibleForTesting;
+import com.google.common.base.Verify;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.Uninterruptibles;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeoutException;
+import okhttp3.RequestBody;
+import okio.Buffer;
+import okio.BufferedSink;
+import okio.Okio;
+import org.chromium.net.UploadDataProvider;
+import org.chromium.net.UploadDataSink;
+
+final class RequestBodyConverterImpl implements RequestBodyConverter {
+
+  private static final long IN_MEMORY_BODY_LENGTH_THRESHOLD_BYTES = 1024 * 1024;
+
+  private final InMemoryRequestBodyConverter inMemoryRequestBodyConverter;
+  private final StreamingRequestBodyConverter streamingRequestBodyConverter;
+
+  RequestBodyConverterImpl(
+      InMemoryRequestBodyConverter inMemoryConverter,
+      StreamingRequestBodyConverter streamingConverter) {
+    this.inMemoryRequestBodyConverter = inMemoryConverter;
+    this.streamingRequestBodyConverter = streamingConverter;
+  }
+
+  static RequestBodyConverterImpl create(ExecutorService bodyReaderExecutor) {
+    return new RequestBodyConverterImpl(
+        new InMemoryRequestBodyConverter(), new StreamingRequestBodyConverter(bodyReaderExecutor));
+  }
+
+  @Override
+  public UploadDataProvider convertRequestBody(RequestBody requestBody, int writeTimeoutMillis)
+      throws IOException {
+    long contentLength = requestBody.contentLength();
+    if (contentLength == -1 || contentLength > IN_MEMORY_BODY_LENGTH_THRESHOLD_BYTES) {
+      return streamingRequestBodyConverter.convertRequestBody(requestBody, writeTimeoutMillis);
+    } else {
+      return inMemoryRequestBodyConverter.convertRequestBody(requestBody, writeTimeoutMillis);
+    }
+  }
+
+  /**
+   * Implementation of {@link RequestBodyConverter} that doesn't need to hold the entire request
+   * body in memory.
+   *
+   * <p>
+   *
+   * <ol>
+   *   <li>{@link RequestBody#writeTo(BufferedSink)} is invoked on the body, but the sink doesn't
+   *       accept any data
+   *   <li>A call to {@link UploadDataProvider#read(UploadDataSink, ByteBuffer)} unblocks the sink,
+   *       which accepts a part of the body (size depends on the buffer's capacity), then blocks
+   *       again. Buffer is sent to Cronet.
+   * </ol>
+   *
+   * This is repeated until the entire body has been read.
+   */
+  @VisibleForTesting
+  static final class StreamingRequestBodyConverter implements RequestBodyConverter {
+
+    private final ExecutorService readerExecutor;
+
+    StreamingRequestBodyConverter(ExecutorService readerExecutor) {
+      this.readerExecutor = readerExecutor;
+    }
+
+    @Override
+    public UploadDataProvider convertRequestBody(RequestBody requestBody, int writeTimeoutMillis) {
+      return new StreamingUploadDataProvider(
+          requestBody, new UploadBodyDataBroker(), readerExecutor, writeTimeoutMillis);
+    }
+
+    private static class StreamingUploadDataProvider extends UploadDataProvider {
+      private final RequestBody okHttpRequestBody;
+      private final UploadBodyDataBroker broker;
+      private final ListeningExecutorService readTaskExecutor;
+      private final long writeTimeoutMillis;
+
+      /** The future for the task that reads the OkHttp request body in the background. */
+      private ListenableFuture<?> readTaskFuture;
+
+      /** The number of bytes we read from the OkHttp body thus far. */
+      private long totalBytesReadFromOkHttp;
+
+      private StreamingUploadDataProvider(
+          RequestBody okHttpRequestBody,
+          UploadBodyDataBroker broker,
+          ExecutorService readTaskExecutor,
+          long writeTimeoutMillis) {
+        this.okHttpRequestBody = okHttpRequestBody;
+        this.broker = broker;
+        if (readTaskExecutor instanceof ListeningExecutorService) {
+          this.readTaskExecutor = (ListeningExecutorService) readTaskExecutor;
+        } else {
+          this.readTaskExecutor = MoreExecutors.listeningDecorator(readTaskExecutor);
+        }
+
+        // So that we don't have to special case infinity. Int.MAX_VALUE is ~infinity for all
+        // practical use cases.
+        this.writeTimeoutMillis = writeTimeoutMillis == 0 ? Integer.MAX_VALUE : writeTimeoutMillis;
+      }
+
+      @Override
+      public long getLength() throws IOException {
+        return okHttpRequestBody.contentLength();
+      }
+
+      @Override
+      public void read(UploadDataSink uploadDataSink, ByteBuffer byteBuffer) throws IOException {
+        ensureReadTaskStarted();
+
+        if (getLength() == -1) {
+          readUnknownBodyLength(uploadDataSink, byteBuffer);
+        } else {
+          readKnownBodyLength(uploadDataSink, byteBuffer);
+        }
+      }
+
+      private void readKnownBodyLength(UploadDataSink uploadDataSink, ByteBuffer byteBuffer)
+          throws IOException {
+        try {
+          UploadBodyDataBroker.ReadResult readResult = readFromOkHttp(byteBuffer);
+
+          if (totalBytesReadFromOkHttp > getLength()) {
+            throw prepareBodyTooLongException(getLength(), totalBytesReadFromOkHttp);
+          }
+
+          if (totalBytesReadFromOkHttp < getLength()) {
+            switch (readResult) {
+              case SUCCESS:
+                uploadDataSink.onReadSucceeded(false);
+                break;
+              case END_OF_BODY:
+                throw new IOException("The source has been exhausted but we expected more data!");
+            }
+            return;
+          }
+          // Else we're handling what's supposed to be the last chunk
+          handleLastBodyRead(uploadDataSink, byteBuffer);
+
+        } catch (TimeoutException | ExecutionException e) {
+          readTaskFuture.cancel(true);
+          uploadDataSink.onReadError(new IOException(e));
+        }
+      }
+
+      /**
+       * The last body read is special for fixed length bodies - if Cronet receives exactly the
+       * right amount of data it won't ask for more, even if there is more data in the stream. As a
+       * result, when we read the advertised number of bytes, we need to make sure that the stream
+       * is indeed finished.
+       */
+      private void handleLastBodyRead(UploadDataSink uploadDataSink, ByteBuffer filledByteBuffer)
+          throws IOException, TimeoutException, ExecutionException {
+        // We reuse the same buffer for the END_OF_DATA read (it should be non-destructive and if
+        // it overwrites what's in there we don't mind as that's an error anyway). We just need
+        // to make sure we restore the original position afterwards. We don't use mark() / reset()
+        // as the mark position can be invalidated by limit manipulation.
+        int bufferPosition = filledByteBuffer.position();
+        filledByteBuffer.position(0);
+
+        UploadBodyDataBroker.ReadResult readResult = readFromOkHttp(filledByteBuffer);
+
+        if (!readResult.equals(UploadBodyDataBroker.ReadResult.END_OF_BODY)) {
+          throw prepareBodyTooLongException(getLength(), totalBytesReadFromOkHttp);
+        }
+
+        Verify.verify(
+            filledByteBuffer.position() == 0,
+            "END_OF_BODY reads shouldn't write anything to the buffer");
+
+        // revert the position change
+        filledByteBuffer.position(bufferPosition);
+
+        uploadDataSink.onReadSucceeded(false);
+      }
+
+      private void readUnknownBodyLength(UploadDataSink uploadDataSink, ByteBuffer byteBuffer) {
+        try {
+          UploadBodyDataBroker.ReadResult readResult = readFromOkHttp(byteBuffer);
+          uploadDataSink.onReadSucceeded(readResult.equals(UploadBodyDataBroker.ReadResult.END_OF_BODY));
+        } catch (TimeoutException | ExecutionException e) {
+          readTaskFuture.cancel(true);
+          uploadDataSink.onReadError(new IOException(e));
+        }
+      }
+
+      private void ensureReadTaskStarted() {
+        // We don't expect concurrent calls so a simple flag is sufficient
+        if (readTaskFuture == null) {
+          readTaskFuture =
+              readTaskExecutor.submit(
+                  (Callable<Void>)
+                      () -> {
+                        BufferedSink bufferedSink = Okio.buffer(broker);
+                        okHttpRequestBody.writeTo(bufferedSink);
+                        bufferedSink.flush();
+                        broker.handleEndOfStreamSignal();
+                        return null;
+                      });
+
+          Futures.addCallback(
+              readTaskFuture,
+              new FutureCallback<Object>() {
+                @Override
+                public void onSuccess(Object result) {}
+
+                @Override
+                public void onFailure(Throwable t) {
+                  broker.setBackgroundReadError(t);
+                }
+              },
+              MoreExecutors.directExecutor());
+        }
+      }
+
+      private UploadBodyDataBroker.ReadResult readFromOkHttp(ByteBuffer byteBuffer)
+          throws TimeoutException, ExecutionException {
+        int positionBeforeRead = byteBuffer.position();
+        UploadBodyDataBroker.ReadResult readResult =
+            Uninterruptibles.getUninterruptibly(
+                broker.enqueueBodyRead(byteBuffer), writeTimeoutMillis, MILLISECONDS);
+        int bytesRead = byteBuffer.position() - positionBeforeRead;
+        totalBytesReadFromOkHttp += bytesRead;
+        return readResult;
+      }
+
+      private static IOException prepareBodyTooLongException(
+          long expectedLength, long minActualLength) {
+        return new IOException(
+            "Expected " + expectedLength + " bytes but got at least " + minActualLength);
+      }
+
+      @Override
+      public void rewind(UploadDataSink uploadDataSink) {
+        // TODO(danstahr): OkHttp 4 can use isOneShot flag here and rewind safely.
+        uploadDataSink.onRewindError(new UnsupportedOperationException("Rewind is not supported!"));
+      }
+    }
+  }
+
+  /**
+   * Converts OkHttp's {@link RequestBody} to Cronet's {@link UploadDataProvider} by materializing
+   * the body in memory first.
+   *
+   * <p>This strategy shouldn't be used for large requests (and for requests with uncapped length)
+   * to avoid OOM issues.
+   */
+  @VisibleForTesting
+  static final class InMemoryRequestBodyConverter implements RequestBodyConverter {
+
+    @Override
+    public UploadDataProvider convertRequestBody(RequestBody requestBody, int writeTimeoutMillis)
+        throws IOException {
+
+      // content length is immutable by contract
+      long length = requestBody.contentLength();
+      if (length < 0 || length > IN_MEMORY_BODY_LENGTH_THRESHOLD_BYTES) {
+        throw new IOException(
+            "Expected definite length less than "
+                + IN_MEMORY_BODY_LENGTH_THRESHOLD_BYTES
+                + "but got "
+                + length);
+      }
+
+      return new UploadDataProvider() {
+        private volatile boolean isMaterialized = false;
+        private final Buffer materializedBody = new Buffer();
+
+        @Override
+        public long getLength() {
+          return length;
+        }
+
+        @Override
+        public void read(UploadDataSink uploadDataSink, ByteBuffer byteBuffer) throws IOException {
+          // We're not expecting any concurrent calls here so a simple flag should be sufficient.
+          if (!isMaterialized) {
+            requestBody.writeTo(materializedBody);
+            materializedBody.flush();
+            isMaterialized = true;
+            long reportedLength = getLength();
+            long actualLength = materializedBody.size();
+            if (actualLength != reportedLength) {
+              throw new IOException(
+                  "Expected " + reportedLength + " bytes but got " + actualLength);
+            }
+          }
+          if (materializedBody.read(byteBuffer) == -1) {
+            // This should never happen - for known body length we shouldn't be called at all
+            // if there's no more data to read.
+            throw new IllegalStateException("The source has been exhausted but we expected more!");
+          }
+          uploadDataSink.onReadSucceeded(false);
+        }
+
+        @Override
+        public void rewind(UploadDataSink uploadDataSink) {
+          // TODO(danstahr): OkHttp 4 can use isOneShot flag here and rewind safely.
+          uploadDataSink.onRewindError(new UnsupportedOperationException());
+        }
+      };
+    }
+  }
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestResponseConverter.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import java.io.IOException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import org.chromium.net.CronetEngine;
+import org.chromium.net.UrlRequest;
+
+/** Converts OkHttp requests to Cronet requests. */
+public final class RequestResponseConverter {
+  private static final String CONTENT_LENGTH_HEADER_NAME = "Content-Length";
+  private static final String CONTENT_TYPE_HEADER_NAME = "Content-Type";
+  private static final String CONTENT_TYPE_HEADER_DEFAULT_VALUE = "application/octet-stream";
+
+  private final CronetEngine cronetEngine;
+  private final Executor uploadDataProviderExecutor;
+  private final ResponseConverter responseConverter;
+  private final RequestBodyConverter requestBodyConverter;
+  private final RedirectStrategy redirectStrategy;
+
+  RequestResponseConverter(
+      CronetEngine cronetEngine,
+      Executor uploadDataProviderExecutor,
+      RequestBodyConverter requestBodyConverter,
+      ResponseConverter responseConverter,
+      RedirectStrategy redirectStrategy) {
+    this.cronetEngine = cronetEngine;
+    this.uploadDataProviderExecutor = uploadDataProviderExecutor;
+    this.requestBodyConverter = requestBodyConverter;
+    this.responseConverter = responseConverter;
+    this.redirectStrategy = redirectStrategy;
+  }
+
+  /**
+   * Converts OkHttp's {@link Request} to a corresponding Cronet's {@link UrlRequest}.
+   *
+   * <p>Since Cronet doesn't have a notion of a Response, which is handled entirely from the
+   * callbacks, this method also returns a {@link Future} like object the
+   * caller should use to obtain the matching {@link Response} for the given request. For example:
+   *
+   * <pre>
+   *   RequestResponseConverter converter = ...
+   *   CronetRequestAndOkHttpResponse reqResp = converter.convert(okHttpRequest);
+   *   reqResp.getRequest.start();
+   *
+   *   // Will block until status code, headers... are available
+   *   Response okHttpResponse = reqResp.getResponse();
+   *
+   *   // use OkHttp Response as usual
+   * </pre>
+   */
+  CronetRequestAndOkHttpResponse convert(
+      Request okHttpRequest, int readTimeoutMillis, int writeTimeoutMillis) throws IOException {
+
+    OkHttpBridgeRequestCallback callback =
+        new OkHttpBridgeRequestCallback(readTimeoutMillis, redirectStrategy);
+
+    // The OkHttp request callback methods are lightweight, the heavy lifting is done by OkHttp /
+    // app owned threads. Use a direct executor to avoid extra thread hops.
+    UrlRequest.Builder builder =
+        cronetEngine
+            .newUrlRequestBuilder(
+                okHttpRequest.url().toString(), callback, MoreExecutors.directExecutor())
+            .allowDirectExecutor();
+
+    builder.setHttpMethod(okHttpRequest.method());
+
+    for (int i = 0; i < okHttpRequest.headers().size(); i++) {
+      builder.addHeader(okHttpRequest.headers().name(i), okHttpRequest.headers().value(i));
+    }
+
+    RequestBody body = okHttpRequest.body();
+
+    if (body != null) {
+      if (okHttpRequest.header(CONTENT_LENGTH_HEADER_NAME) == null && body.contentLength() != -1) {
+        builder.addHeader(CONTENT_LENGTH_HEADER_NAME, String.valueOf(body.contentLength()));
+      }
+
+      if (body.contentLength() != 0) {
+        if (body.contentType() != null) {
+          builder.addHeader(CONTENT_TYPE_HEADER_NAME, body.contentType().toString());
+        } else if (okHttpRequest.header(CONTENT_TYPE_HEADER_NAME) == null) {
+          // Cronet always requires content-type to be present when a body is present. Use a generic
+          // value if one isn't provided.
+          builder.addHeader(CONTENT_TYPE_HEADER_NAME, CONTENT_TYPE_HEADER_DEFAULT_VALUE);
+        } // else use the header
+
+        builder.setUploadDataProvider(
+            requestBodyConverter.convertRequestBody(body, writeTimeoutMillis),
+            uploadDataProviderExecutor);
+      }
+    }
+
+    return new CronetRequestAndOkHttpResponse(
+        builder.build(), createResponseSupplier(okHttpRequest, callback));
+  }
+
+  private ResponseSupplier createResponseSupplier(
+      Request request, OkHttpBridgeRequestCallback callback) {
+    return new ResponseSupplier() {
+      @Override
+      public Response getResponse() throws IOException {
+        return responseConverter.toResponse(request, callback);
+      }
+
+      @Override
+      public ListenableFuture<Response> getResponseFuture() {
+        return responseConverter.toResponseAsync(request, callback);
+      }
+    };
+  }
+
+  /** A {@link Future} like holder for OkHttp's {@link Response}. */
+  private interface ResponseSupplier {
+    Response getResponse() throws IOException;
+
+    ListenableFuture<Response> getResponseFuture();
+  }
+
+  /** A simple data class for bundling Cronet request and OkHttp response. */
+  static final class CronetRequestAndOkHttpResponse {
+    private final UrlRequest request;
+    private final ResponseSupplier responseSupplier;
+
+    CronetRequestAndOkHttpResponse(UrlRequest request, ResponseSupplier responseSupplier) {
+      this.request = request;
+      this.responseSupplier = responseSupplier;
+    }
+
+    public UrlRequest getRequest() {
+      return request;
+    }
+
+    public Response getResponse() throws IOException {
+      return responseSupplier.getResponse();
+    }
+
+    public ListenableFuture<Response> getResponseAsync() {
+      return responseSupplier.getResponseFuture();
+    }
+  }
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestResponseConverterBasedBuilder.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import org.chromium.net.CronetEngine;
+import org.chromium.net.UploadDataProvider;
+
+abstract class RequestResponseConverterBasedBuilder<
+    SubBuilderT extends RequestResponseConverterBasedBuilder<?, ? extends ObjectBeingBuiltT>,
+    ObjectBeingBuiltT> {
+  private static final int DEFAULT_THREAD_POOL_SIZE = 4;
+
+  private final CronetEngine cronetEngine;
+  private int uploadDataProviderExecutorSize = DEFAULT_THREAD_POOL_SIZE;
+  // Not setting the default straight away to lazy initialize the object if it ends up not being
+  // used.
+  private RedirectStrategy redirectStrategy = null;
+  private final SubBuilderT castedThis;
+
+  @SuppressWarnings("unchecked") // checked as a precondition
+  RequestResponseConverterBasedBuilder(CronetEngine cronetEngine, Class<SubBuilderT> clazz) {
+    this.cronetEngine = checkNotNull(cronetEngine);
+    checkArgument(this.getClass().equals(clazz));
+    castedThis = (SubBuilderT) this;
+  }
+
+  /**
+   * Sets the size of upload data provider executor. The same executor is used for all upload data
+   * providers within the interceptor.
+   *
+   * @see org.chromium.net.UrlRequest.Builder#setUploadDataProvider(UploadDataProvider, Executor)
+   */
+  public final SubBuilderT setUploadDataProviderExecutorSize(int size) {
+    checkArgument(size > 0, "The number of threads must be positive!");
+    uploadDataProviderExecutorSize = size;
+    return castedThis;
+  }
+
+  /**
+   * Sets the strategy for following redirects.
+   *
+   * <p>Note that the Cronet (i.e. Chromium) wide safeguards will still apply if one attempts to
+   * follow redirects too many times.
+   */
+  public final SubBuilderT setRedirectStrategy(RedirectStrategy redirectStrategy) {
+    checkNotNull(redirectStrategy);
+    this.redirectStrategy = redirectStrategy;
+    return castedThis;
+  }
+
+  abstract ObjectBeingBuiltT build(RequestResponseConverter converter);
+
+  public final ObjectBeingBuiltT build() {
+    if (redirectStrategy == null) {
+      redirectStrategy = RedirectStrategy.defaultStrategy();
+    }
+
+    RequestResponseConverter converter =
+        new RequestResponseConverter(
+            cronetEngine,
+            Executors.newFixedThreadPool(uploadDataProviderExecutorSize),
+            // There must always be enough executors to blocking-read the OkHttp request bodies
+            // otherwise deadlocks can occur.
+            RequestBodyConverterImpl.create(Executors.newCachedThreadPool()),
+            new ResponseConverter(),
+            redirectStrategy);
+
+    return build(converter);
+  }
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/ResponseConverter.java

@@ -0,0 +1,264 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import androidx.annotation.NonNull;
+import androidx.annotation.Nullable;
+
+import com.google.common.base.Ascii;
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.Uninterruptibles;
+import java.io.IOException;
+import java.net.ProtocolException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import okhttp3.MediaType;
+import okhttp3.Protocol;
+import okhttp3.Request;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import okio.Okio;
+import okio.Source;
+import org.chromium.net.UrlResponseInfo;
+
+/**
+ * Converts Cronet's responses (or, more precisely, its chunks as they come from Cronet's {@link
+ * org.chromium.net.UrlRequest.Callback}), to OkHttp's {@link Response}.
+ */
+final class ResponseConverter {
+  private static final String CONTENT_LENGTH_HEADER_NAME = "Content-Length";
+  private static final String CONTENT_TYPE_HEADER_NAME = "Content-Type";
+  private static final String CONTENT_ENCODING_HEADER_NAME = "Content-Encoding";
+
+  // https://source.chromium.org/search?q=symbol:FilterSourceStream::ParseEncodingType%20f:cc
+  private static final ImmutableSet<String> ENCODINGS_HANDLED_BY_CRONET =
+      ImmutableSet.of("br", "deflate", "gzip", "x-gzip");
+
+  private static final Splitter COMMA_SPLITTER = Splitter.on(',').trimResults().omitEmptyStrings();
+
+  /**
+   * Creates an OkHttp's Response from the OkHttp-Cronet bridging callback.
+   *
+   * <p>As long as the callback's {@code UrlResponseInfo} is available this method is non-blocking.
+   * However, this method doesn't fetch the entire body response. As a result, subsequent calls to
+   * the result's {@link Response#body()} methods might block further.
+   */
+  Response toResponse(Request request, OkHttpBridgeRequestCallback callback) throws IOException {
+    UrlResponseInfo cronetResponseInfo = getFutureValue(callback.getUrlResponseInfo());
+    Response.Builder responseBuilder =
+        createResponse(request, cronetResponseInfo, getFutureValue(callback.getBodySource()));
+
+    List<UrlResponseInfo> redirectResponseInfos = callback.getUrlResponseInfoChain();
+    List<String> urlChain = cronetResponseInfo.getUrlChain();
+
+    if (!redirectResponseInfos.isEmpty()) {
+      checkArgument(
+          urlChain.size() == redirectResponseInfos.size() + 1,
+          "The number of redirects should be consistent across URLs and headers!");
+
+      Response priorResponse = null;
+      for (int i = 0; i < redirectResponseInfos.size(); i++) {
+        Request redirectedRequest = request.newBuilder().url(urlChain.get(i)).build();
+        priorResponse =
+            createResponse(redirectedRequest, redirectResponseInfos.get(i), null)
+                .priorResponse(priorResponse)
+                .build();
+      }
+
+      responseBuilder
+          .request(request.newBuilder().url(Iterables.getLast(urlChain)).build())
+          .priorResponse(priorResponse);
+    }
+
+    return responseBuilder.build();
+  }
+
+  ListenableFuture<Response> toResponseAsync(
+      Request request, OkHttpBridgeRequestCallback callback) {
+    return Futures.whenAllComplete(callback.getUrlResponseInfo(), callback.getBodySource())
+        .call(() -> toResponse(request, callback), MoreExecutors.directExecutor());
+  }
+
+  private static Response.Builder createResponse(
+      Request request, UrlResponseInfo cronetResponseInfo, @Nullable Source bodySource)
+      throws IOException {
+
+    Response.Builder responseBuilder = new Response.Builder();
+
+    @Nullable String contentType = getLastHeaderValue(CONTENT_TYPE_HEADER_NAME, cronetResponseInfo);
+
+    // If all content encodings are those known to Cronet natively, Cronet decodes the body stream.
+    // Otherwise, it's sent to the callbacks verbatim. For consistency with OkHttp, we only leave
+    // the Content-Encoding headers if Cronet didn't decode the request. Similarly, for consistency,
+    // we strip the Content-Length header of decoded responses.
+
+    @Nullable String contentLengthString = null;
+
+    // Theoretically, the content encodings can be scattered across multiple comma separated
+    // Content-Encoding headers. This list contains individual encodings.
+    List<String> contentEncodingItems = new ArrayList<>();
+
+    for (String contentEncodingHeaderValue :
+        getOrDefault(
+            cronetResponseInfo.getAllHeaders(),
+            CONTENT_ENCODING_HEADER_NAME,
+            Collections.emptyList())) {
+      Iterables.addAll(contentEncodingItems, COMMA_SPLITTER.split(contentEncodingHeaderValue));
+    }
+
+    boolean keepEncodingAffectedHeaders =
+        contentEncodingItems.isEmpty()
+            || !ENCODINGS_HANDLED_BY_CRONET.containsAll(contentEncodingItems);
+
+    if (keepEncodingAffectedHeaders) {
+      contentLengthString = getLastHeaderValue(CONTENT_LENGTH_HEADER_NAME, cronetResponseInfo);
+    }
+
+    ResponseBody responseBody = null;
+    if (bodySource != null) {
+      responseBody =
+          createResponseBody(
+              request,
+              cronetResponseInfo.getHttpStatusCode(),
+              contentType,
+              contentLengthString,
+              bodySource);
+    }
+
+    responseBuilder
+        .request(request)
+        .code(cronetResponseInfo.getHttpStatusCode())
+        .message(cronetResponseInfo.getHttpStatusText())
+        .protocol(convertProtocol(cronetResponseInfo.getNegotiatedProtocol()))
+        .body(responseBody);
+
+    for (Map.Entry<String, String> header : cronetResponseInfo.getAllHeadersAsList()) {
+      boolean copyHeader = true;
+      if (!keepEncodingAffectedHeaders) {
+        if (Ascii.equalsIgnoreCase(header.getKey(), CONTENT_LENGTH_HEADER_NAME)
+            || Ascii.equalsIgnoreCase(header.getKey(), CONTENT_ENCODING_HEADER_NAME)) {
+          copyHeader = false;
+        }
+      }
+      if (copyHeader) {
+        responseBuilder.addHeader(header.getKey(), header.getValue());
+      }
+    }
+
+    return responseBuilder;
+  }
+
+  /**
+   * Creates an OkHttp's ResponseBody from the OkHttp-Cronet bridging callback.
+   *
+   * <p>As long as the callback's {@code UrlResponseInfo} is available this method is non-blocking.
+   * However, this method doesn't fetch the entire body response. As a result, subsequent calls to
+   * {@link ResponseBody} methods might block further to fetch parts of the body.
+   */
+  private static ResponseBody createResponseBody(
+      Request request,
+      int httpStatusCode,
+      @Nullable String contentType,
+      @Nullable String contentLengthString,
+      Source bodySource)
+      throws IOException {
+
+    long contentLength;
+
+    // Ignore content-length header for HEAD requests (consistency with OkHttp)
+    if (request.method().equals("HEAD")) {
+      contentLength = 0;
+    } else {
+      try {
+        contentLength = contentLengthString != null ? Long.parseLong(contentLengthString) : -1;
+      } catch (NumberFormatException e) {
+        // TODO(danstahr): add logging
+        contentLength = -1;
+      }
+    }
+
+    // Check for absence of body in No Content / Reset Content responses (OkHttp consistency)
+    if ((httpStatusCode == 204 || httpStatusCode == 205) && contentLength > 0) {
+      throw new ProtocolException(
+          "HTTP " + httpStatusCode + " had non-zero Content-Length: " + contentLengthString);
+    }
+
+    return ResponseBody.create(
+        contentType != null ? MediaType.parse(contentType) : null,
+        contentLength,
+        Okio.buffer(bodySource));
+  }
+
+  /** Converts Cronet's negotiated protocol string to OkHttp's {@link Protocol}. */
+  private static Protocol convertProtocol(String negotiatedProtocol) {
+    // See
+    // https://www.iana.org/assignments/tls-extensiontype-values/tls-extensiontype-values.xhtml#alpn-protocol-ids
+    if (negotiatedProtocol.contains("quic")) {
+      return Protocol.QUIC;
+    } else if (negotiatedProtocol.contains("h3")) {
+      // TODO(danstahr): Should be h3 for newer OkHttp
+      return Protocol.QUIC;
+    } else if (negotiatedProtocol.contains("spdy")) {
+      return Protocol.HTTP_2;
+    } else if (negotiatedProtocol.contains("h2")) {
+      return Protocol.HTTP_2;
+    } else if (negotiatedProtocol.contains("http/1.1")) {
+      return Protocol.HTTP_1_1;
+    }
+
+    return Protocol.HTTP_1_0;
+  }
+
+  /** Returns the last header value for the given name, or null if the header isn't present. */
+  @Nullable
+  private static String getLastHeaderValue(String name, UrlResponseInfo responseInfo) {
+    List<String> headers = responseInfo.getAllHeaders().get(name);
+    if (headers == null || headers.isEmpty()) {
+      return null;
+    }
+    return Iterables.getLast(headers);
+  }
+
+  private static <T> T getFutureValue(Future<T> future) throws IOException {
+    try {
+      return Uninterruptibles.getUninterruptibly(future);
+    } catch (ExecutionException e) {
+      throw new IOException(e);
+    }
+  }
+
+  private static <K, V> V getOrDefault(Map<K, V> map, K key, @NonNull V defaultValue) {
+    V value = map.get(key);
+    if (value == null) {
+      return checkNotNull(defaultValue);
+    } else {
+      return value;
+    }
+  }
+}

foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/UploadBodyDataBroker.java

@@ -0,0 +1,177 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * 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.mogo.cloud.network.cronet;
+
+import static com.google.common.base.Preconditions.checkState;
+
+import android.util.Pair;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.SettableFuture;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import okio.Buffer;
+import okio.Sink;
+import okio.Timeout;
+import org.chromium.net.UploadDataSink;
+
+final class UploadBodyDataBroker implements Sink {
+
+  /**
+   * The read request calls to {@link org.chromium.net.UploadDataProvider#read(UploadDataSink,
+   * ByteBuffer)} associated with this broker that we haven't started handling.
+   *
+   * <p>We don't expect more than one parallel read call for a single request body provider.
+   */
+  private final BlockingQueue<Pair<ByteBuffer, SettableFuture<ReadResult>>> pendingRead =
+      new ArrayBlockingQueue<>(1);
+
+  /**
+   * Whether the sink has been closed.
+   *
+   * <p>Calling close() has no practical use but we check that nobody tries to write to the sink
+   * after closing it, which is an indication of misuse.
+   */
+  private final AtomicBoolean isClosed = new AtomicBoolean();
+
+  /**
+   * The exception thrown by the body reading background thread, if any. The exception will be
+   * rethrown every time someone attempts to continue reading the body.
+   */
+  private final AtomicReference<Throwable> backgroundReadThrowable = new AtomicReference<>();
+
+  /**
+   * Indicates that Cronet is ready to receive another body part.
+   *
+   * <p>This method is executed by Cronet's upload data provider.
+   */
+  Future<ReadResult> enqueueBodyRead(ByteBuffer readBuffer) {
+    Throwable backgroundThrowable = backgroundReadThrowable.get();
+    if (backgroundThrowable != null) {
+      return Futures.immediateFailedFuture(backgroundThrowable);
+    }
+    SettableFuture<ReadResult> future = SettableFuture.create();
+    pendingRead.add(Pair.create(readBuffer, future));
+
+    // Properly handle interleaving handleBackgroundReadError / enqueueBodyRead calls.
+    if ((backgroundThrowable = backgroundReadThrowable.get()) != null) {
+      future.setException(backgroundThrowable);
+    }
+    return future;
+  }
+
+  /**
+   * Signals that reading the OkHttp body failed with the given throwable.
+   *
+   * <p>This method is executed by the background OkHttp body reading thread.
+   */
+  void setBackgroundReadError(Throwable t) {
+    backgroundReadThrowable.set(t);
+    Pair<ByteBuffer, SettableFuture<ReadResult>> read = pendingRead.poll();
+    if (read != null) {
+      read.second.setException(t);
+    }
+  }
+
+  /**
+   * Signals that reading the body has ended and no future bytes will be sent.
+   *
+   * <p>This method is executed by the background OkHttp body reading thread.
+   */
+  void handleEndOfStreamSignal() throws IOException {
+    if (isClosed.getAndSet(true)) {
+      throw new IllegalStateException("Already closed");
+    }
+
+    getPendingCronetRead().second.set(ReadResult.END_OF_BODY);
+  }
+
+  /**
+   * {@inheritDoc}
+   *
+   * <p>This method is executed by the background OkHttp body reading thread.
+   */
+  @Override
+  public void write(Buffer source, long byteCount) throws IOException {
+    // This is just a safeguard, close() is a no-op if the body length contract is honored.
+    checkState(!isClosed.get());
+
+    long bytesRemaining = byteCount;
+
+    while (bytesRemaining != 0) {
+      Pair<ByteBuffer, SettableFuture<ReadResult>> payload = getPendingCronetRead();
+
+      ByteBuffer readBuffer = payload.first;
+      SettableFuture<ReadResult> future = payload.second;
+
+      int originalBufferLimit = readBuffer.limit();
+      int bytesToDrain = (int) Math.min(originalBufferLimit, bytesRemaining);
+
+      readBuffer.limit(bytesToDrain);
+
+      try {
+        long bytesRead = source.read(readBuffer);
+        if (bytesRead == -1) {
+          IOException e = new IOException("The source has been exhausted but we expected more!");
+          future.setException(e);
+          throw e;
+        }
+        bytesRemaining -= bytesRead;
+        readBuffer.limit(originalBufferLimit);
+        future.set(ReadResult.SUCCESS);
+      } catch (IOException e) {
+        future.setException(e);
+        throw e;
+      }
+    }
+  }
+
+  private Pair<ByteBuffer, SettableFuture<ReadResult>> getPendingCronetRead() throws IOException {
+    try {
+      return pendingRead.take();
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new IOException("Interrupted while waiting for a read to finish!");
+    }
+  }
+
+  @Override
+  public void close() {
+    isClosed.set(true);
+  }
+
+  @Override
+  public void flush() {
+    // Not necessary, we "flush" by sending the data to Cronet straight away when write() is called.
+    // Note that this class is wrapped with a okio buffer so writes to the outer layer won't be
+    // seen by this class immediately.
+  }
+
+  @Override
+  public Timeout timeout() {
+    return Timeout.NONE;
+  }
+
+  enum ReadResult {
+    SUCCESS,
+    END_OF_BODY
+  }
+}

gradle.properties

@@ -36,24 +36,24 @@ PASSWORD=xintai2018
 RELEASE=true
 # AI CLOUD 云平台
 # 工具类
-MOGO_UTILS_VERSION=1.4.7.16
+MOGO_UTILS_VERSION=1.4.7.17
 # 网络请求
-MOGO_NETWORK_VERSION=1.4.7.16
+MOGO_NETWORK_VERSION=1.4.7.17
 # 鉴权
-MOGO_PASSPORT_VERSION=1.4.7.16
+MOGO_PASSPORT_VERSION=1.4.7.17
 # 常链接
-MOGO_SOCKET_VERSION=1.4.7.16
+MOGO_SOCKET_VERSION=1.4.7.17
 # 数据采集
-MOGO_REALTIME_VERSION=1.4.7.16
+MOGO_REALTIME_VERSION=1.4.7.17
 # 探路，道路事件发布，获取
-MOGO_TANLU_VERSION=1.4.7.16
+MOGO_TANLU_VERSION=1.4.7.17
 # 直播推流
-MOGO_LIVE_VERSION=1.4.7.16
+MOGO_LIVE_VERSION=1.4.7.17
 # 直播拉流
-MOGO_TRAFFICLIVE_VERSION=1.4.7.16
+MOGO_TRAFFICLIVE_VERSION=1.4.7.17
 # 定位服务
-MOGO_LOCATION_VERSION=1.4.7.16
+MOGO_LOCATION_VERSION=1.4.7.17
 # 远程通讯模块
-MOGO_TELEMATIC_VERSION=1.4.7.16
+MOGO_TELEMATIC_VERSION=1.4.7.17
 # v2x
-MOGO_V2X_VERSION=1.4.7.16
+MOGO_V2X_VERSION=1.4.7.17