diff --git a/foudations/mogo-network/build.gradle b/foudations/mogo-network/build.gradle index eea7a1c..916e531 100644 --- a/foudations/mogo-network/build.gradle +++ b/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}" diff --git a/foudations/mogo-network/libs/cronet_api.jar b/foudations/mogo-network/libs/cronet_api.jar new file mode 100644 index 0000000..ae3a4f2 Binary files /dev/null and b/foudations/mogo-network/libs/cronet_api.jar differ diff --git a/foudations/mogo-network/libs/cronet_impl_common_java.jar b/foudations/mogo-network/libs/cronet_impl_common_java.jar new file mode 100644 index 0000000..aba2420 Binary files /dev/null and b/foudations/mogo-network/libs/cronet_impl_common_java.jar differ diff --git a/foudations/mogo-network/libs/cronet_impl_native_java.jar b/foudations/mogo-network/libs/cronet_impl_native_java.jar new file mode 100644 index 0000000..8e05a39 Binary files /dev/null and b/foudations/mogo-network/libs/cronet_impl_native_java.jar differ diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/OkHttpFactory.kt b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/OkHttpFactory.kt index fcc6bac..ab17f08 100644 --- a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/OkHttpFactory.kt +++ b/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) diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetCallFactory.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetCallFactory.java new file mode 100644 index 0000000..606574e --- /dev/null +++ b/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 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() { + @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 { + 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); + } + } +} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetInterceptor.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetInterceptor.java new file mode 100644 index 0000000..83b0dc4 --- /dev/null +++ b/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. + * + *

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. + * + *

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: + * + *

    + *
  1. The entirety of OkHttp core is bypassed. This includes caching configuration and network + * interceptors. + *
  2. Some response fields are not being populated due to mismatches between Cronet's and + * OkHttp's architecture. TODO(danstahr): add a concrete list). + *
+ */ +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 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> activeCallsIterator = + activeCalls.entrySet().iterator(); + + while (activeCallsIterator.hasNext()) { + try { + Entry 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); + } + } +} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetTimeoutException.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetTimeoutException.java new file mode 100644 index 0000000..7876212 --- /dev/null +++ b/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 {} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetTransportResponseBody.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/CronetTransportResponseBody.java new file mode 100644 index 0000000..1db0303 --- /dev/null +++ b/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(); +} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/OkHttpBridgeRequestCallback.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/OkHttpBridgeRequestCallback.java new file mode 100644 index 0000000..a9eec34 --- /dev/null +++ b/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. + * + *

Translating the UrlResponseInfo is relatively straightforward as the entire object is + * available immediately and is relatively small, so it can easily fit in memory. + * + *

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 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}. + * + *

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 callbackResults = new ArrayBlockingQueue<>(2); + + /** The response headers. */ + private final SettableFuture 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 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 getUrlResponseInfo() { + return headersFuture; + } + + /** + * Returns the OkHttp {@link Source} for the request associated with this callback. + * + *

Note that retrieving data from the {@code Source} instance might block further as the + * response body is streamed. + */ + ListenableFuture getBodySource() { + return bodySourceFuture; + } + + List 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 + } +} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/QuicConfig.kt b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/QuicConfig.kt new file mode 100644 index 0000000..c9afd90 --- /dev/null +++ b/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 + } +} \ No newline at end of file diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RedirectStrategy.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RedirectStrategy.java new file mode 100644 index 0000000..7cfcdc9 --- /dev/null +++ b/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. + * + *

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() {} +} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestBodyConverter.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestBodyConverter.java new file mode 100644 index 0000000..e09f9d7 --- /dev/null +++ b/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; +} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestBodyConverterImpl.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestBodyConverterImpl.java new file mode 100644 index 0000000..5dd6f93 --- /dev/null +++ b/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. + * + *

+ * + *

    + *
  1. {@link RequestBody#writeTo(BufferedSink)} is invoked on the body, but the sink doesn't + * accept any data + *
  2. 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. + *
+ * + * 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) + () -> { + BufferedSink bufferedSink = Okio.buffer(broker); + okHttpRequestBody.writeTo(bufferedSink); + bufferedSink.flush(); + broker.handleEndOfStreamSignal(); + return null; + }); + + Futures.addCallback( + readTaskFuture, + new FutureCallback() { + @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. + * + *

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()); + } + }; + } + } +} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestResponseConverter.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestResponseConverter.java new file mode 100644 index 0000000..e0d4b8f --- /dev/null +++ b/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}. + * + *

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: + * + * 

+   *   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
+   *
+ */ + 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 getResponseFuture() { + return responseConverter.toResponseAsync(request, callback); + } + }; + } + + /** A {@link Future} like holder for OkHttp's {@link Response}. */ + private interface ResponseSupplier { + Response getResponse() throws IOException; + + ListenableFuture 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 getResponseAsync() { + return responseSupplier.getResponseFuture(); + } + } +} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestResponseConverterBasedBuilder.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/RequestResponseConverterBasedBuilder.java new file mode 100644 index 0000000..f5f3022 --- /dev/null +++ b/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, + 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 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. + * + *

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); + } +} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/ResponseConverter.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/ResponseConverter.java new file mode 100644 index 0000000..ed7861f --- /dev/null +++ b/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 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. + * + *

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 redirectResponseInfos = callback.getUrlResponseInfoChain(); + List 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 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 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 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. + * + *

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 headers = responseInfo.getAllHeaders().get(name); + if (headers == null || headers.isEmpty()) { + return null; + } + return Iterables.getLast(headers); + } + + private static T getFutureValue(Future future) throws IOException { + try { + return Uninterruptibles.getUninterruptibly(future); + } catch (ExecutionException e) { + throw new IOException(e); + } + } + + private static V getOrDefault(Map map, K key, @NonNull V defaultValue) { + V value = map.get(key); + if (value == null) { + return checkNotNull(defaultValue); + } else { + return value; + } + } +} diff --git a/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/UploadBodyDataBroker.java b/foudations/mogo-network/src/main/java/com/mogo/cloud/network/cronet/UploadBodyDataBroker.java new file mode 100644 index 0000000..7223fe7 --- /dev/null +++ b/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. + * + *

We don't expect more than one parallel read call for a single request body provider. + */ + private final BlockingQueue>> pendingRead = + new ArrayBlockingQueue<>(1); + + /** + * Whether the sink has been closed. + * + *

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 backgroundReadThrowable = new AtomicReference<>(); + + /** + * Indicates that Cronet is ready to receive another body part. + * + *

This method is executed by Cronet's upload data provider. + */ + Future enqueueBodyRead(ByteBuffer readBuffer) { + Throwable backgroundThrowable = backgroundReadThrowable.get(); + if (backgroundThrowable != null) { + return Futures.immediateFailedFuture(backgroundThrowable); + } + SettableFuture 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. + * + *

This method is executed by the background OkHttp body reading thread. + */ + void setBackgroundReadError(Throwable t) { + backgroundReadThrowable.set(t); + Pair> read = pendingRead.poll(); + if (read != null) { + read.second.setException(t); + } + } + + /** + * Signals that reading the body has ended and no future bytes will be sent. + * + *

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} + * + *

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> payload = getPendingCronetRead(); + + ByteBuffer readBuffer = payload.first; + SettableFuture 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> 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 + } +} diff --git a/foudations/mogo-network/src/main/jniLibs/arm64-v8a/libcronet.100.0.4845.0.so b/foudations/mogo-network/src/main/jniLibs/arm64-v8a/libcronet.100.0.4845.0.so new file mode 100644 index 0000000..f15977b Binary files /dev/null and b/foudations/mogo-network/src/main/jniLibs/arm64-v8a/libcronet.100.0.4845.0.so differ diff --git a/gradle.properties b/gradle.properties index 8448aa2..5f09890 100644 --- a/gradle.properties +++ b/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