Fix external cancels propagation in rxjava2 and rxjava3 adapters

CHANGELOG.md

@@ -13,7 +13,7 @@
 
 **Fixed**
 
- - Nothing yet!
+ - Propagate timeouts and other external cancels in retrofit-adapter for rxjava2 and rxjava3
 
 
 ## [3.0.0] - 2025-05-15

retrofit-adapters/rxjava2/src/main/java/retrofit2/adapter/rxjava2/CallEnqueueObservable.java

@@ -82,7 +82,7 @@ public void onResponse(Call<T> call, Response<T> response) {
 
     @Override
     public void onFailure(Call<T> call, Throwable t) {
-      if (call.isCanceled()) return;
+      if (disposed) return;
 
       try {
         observer.onError(t);

retrofit-adapters/rxjava2/src/test/java/retrofit2/adapter/rxjava2/CancelDisposeTest.java

@@ -21,7 +21,12 @@
 
 import io.reactivex.Observable;
 import io.reactivex.disposables.Disposable;
+import io.reactivex.observers.TestObserver;
+
+import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
+
 import okhttp3.Call;
 import okhttp3.OkHttpClient;
 import okhttp3.mockwebserver.MockWebServer;
@@ -79,4 +84,14 @@ public void cancelDoesNotDispose() {
     calls.get(0).cancel();
     assertFalse(disposable.isDisposed());
   }
+
+  @Test
+  public void cancelSetsError() throws InterruptedException {
+    TestObserver<?> testObserver = service.go().test();
+    List<Call> calls = client.dispatcher().runningCalls();
+    assertEquals(1, calls.size());
+    calls.get(0).cancel();
+    testObserver.await(10, TimeUnit.SECONDS);
+    testObserver.assertError(e -> e instanceof IOException && "Canceled".equals(e.getMessage()));
+  }
 }

retrofit-adapters/rxjava3/src/main/java/retrofit2/adapter/rxjava3/CallEnqueueObservable.java

@@ -82,7 +82,7 @@ public void onResponse(Call<T> call, Response<T> response) {
 
     @Override
     public void onFailure(Call<T> call, Throwable t) {
-      if (call.isCanceled()) return;
+      if (disposed) return;
 
       try {
         observer.onError(t);

retrofit-adapters/rxjava3/src/test/java/retrofit2/adapter/rxjava3/CancelDisposeTest.java

@@ -21,7 +21,12 @@
 
 import io.reactivex.rxjava3.core.Observable;
 import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.observers.TestObserver;
+
+import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
+
 import okhttp3.Call;
 import okhttp3.OkHttpClient;
 import okhttp3.mockwebserver.MockWebServer;
@@ -79,4 +84,14 @@ public void cancelDoesNotDispose() {
     calls.get(0).cancel();
     assertFalse(disposable.isDisposed());
   }
+
+  @Test
+  public void cancelSetsError() throws InterruptedException {
+    TestObserver<?> testObserver = service.go().test();
+    List<Call> calls = client.dispatcher().runningCalls();
+    assertEquals(1, calls.size());
+    calls.get(0).cancel();
+    testObserver.await(10, TimeUnit.SECONDS);
+    testObserver.assertError(e -> e instanceof IOException && "Canceled".equals(e.getMessage()));
+  }
 }