Update default subscription type and update auth interceptor to not handle all errors (#4)

Author: aleksandar-apostolov

stream-android-core/src/main/java/io/getstream/android/core/api/StreamClient.kt

@@ -259,10 +259,11 @@ fun StreamClient(
             httpBuilder.apply {
                 addInterceptor(StreamOkHttpInterceptors.clientInfo(clientInfoHeader))
                 addInterceptor(StreamOkHttpInterceptors.apiKey(apiKey))
+                addInterceptor(StreamOkHttpInterceptors.connectionId(connectionIdHolder))
                 addInterceptor(
                     StreamOkHttpInterceptors.auth("jwt", tokenManager, compositeSerialization)
                 )
-                addInterceptor(StreamOkHttpInterceptors.connectionId(connectionIdHolder))
+                addInterceptor(StreamOkHttpInterceptors.error(compositeSerialization))
             }
         }
         configuredInterceptors.forEach { httpBuilder.addInterceptor(it) }

stream-android-core/src/main/java/io/getstream/android/core/api/model/config/StreamClientSerializationConfig.kt

@@ -17,8 +17,8 @@ package io.getstream.android.core.api.model.config
 
 import io.getstream.android.core.annotations.StreamCoreApi
 import io.getstream.android.core.api.model.event.StreamClientWsEvent
-import io.getstream.android.core.api.serialization.StreamJsonSerialization
 import io.getstream.android.core.api.serialization.StreamEventSerialization
+import io.getstream.android.core.api.serialization.StreamJsonSerialization
 
 /**
  * Configuration for serialization and deserialization in the Stream client.

stream-android-core/src/main/java/io/getstream/android/core/api/serialization/StreamEventSerialization.kt

@@ -49,10 +49,9 @@ interface StreamEventSerialization<T> {
     fun deserialize(raw: String): Result<T>
 }
 
-
-
 /**
- * Creates a new [StreamEventSerialization] instance that can serialize [io.getstream.android.core.api.model.event.StreamClientWsEvent] objects.
+ * Creates a new [StreamEventSerialization] instance that can serialize
+ * [io.getstream.android.core.api.model.event.StreamClientWsEvent] objects.
  *
  * @param jsonParser The [StreamJsonSerialization] to use for JSON serialization and
  *   deserialization.

stream-android-core/src/main/java/io/getstream/android/core/api/subscribe/StreamSubscriptionManager.kt

@@ -44,7 +44,7 @@ interface StreamSubscriptionManager<T> {
      *
      * @property retention Controls how the manager retains the listener reference.
      */
-    data class Options(val retention: Retention = Retention.AUTO_REMOVE) {
+    data class Options(val retention: Retention = Retention.KEEP_UNTIL_CANCELLED) {
         /** Retention policy for a subscribed listener. */
         enum class Retention {
             /**

stream-android-core/src/main/java/io/getstream/android/core/internal/http/interceptor/StreamAuthInterceptor.kt

@@ -67,25 +67,25 @@ internal class StreamAuthInterceptor(
         val original = chain.request()
         val authed = original.withAuthHeaders(authType, token.rawValue)
 
-        val firstResponse = chain.proceed(authed)
-        if (firstResponse.isSuccessful) {
-            return firstResponse
+        val first = chain.proceed(authed)
+        if (first.isSuccessful) {
+            return first
         }
 
-        val errorBody = firstResponse.peekBody(PEEK_ERROR_BYTES_MAX).string()
-        val parsed = jsonParser.fromJson(errorBody, StreamEndpointErrorData::class.java)
+        // Peek only; do NOT consume
+        val peeked = first.peekBody(PEEK_ERROR_BYTES_MAX).string()
+        val parsed = jsonParser.fromJson(peeked, StreamEndpointErrorData::class.java)
 
-        // Guard against infinite loops: retry at most once per request.
         val alreadyRetried = original.header(HEADER_RETRIED_ON_AUTH) == "present"
 
         if (parsed.isSuccess) {
             val error = parsed.getOrEndpointException("Failed to parse error body.")
-            if (!alreadyRetried && isTokenInvalidErrorCode(error.code)) {
-                // Refresh and retry once.
-                firstResponse.close()
-                tokenManager
-                    .invalidate()
-                    .getOrEndpointException(message = "Failed to invalidate token")
+
+            // Only handle token errors here
+            if (isTokenInvalidErrorCode(error.code) && !alreadyRetried) {
+                // refresh & retry once
+                first.close()
+                tokenManager.invalidate().getOrEndpointException("Failed to invalidate token")
                 val refreshed =
                     runBlocking { tokenManager.refresh() }
                         .getOrEndpointException("Failed to refresh token")
@@ -97,19 +97,15 @@ internal class StreamAuthInterceptor(
                         .header(HEADER_RETRIED_ON_AUTH, "present")
                         .build()
 
-                return chain.proceed(retried)
+                return chain.proceed(retried) // pass result (ok or error) downstream
             }
 
-            // Non-token error or we already retried: surface a structured exception.
-            firstResponse.close()
-            throw StreamEndpointException("Failed request: ${original.url}", error, null)
+            // Non-token error, or token error but we already retried:
+            // pass the original failed response downstream; DO NOT throw here.
+            return first
         } else {
-            // Couldn’t parse error, still fail in a consistent way.
-            firstResponse.close()
-            throw StreamEndpointException(
-                "Failed to serialize response error body: ${original.url}",
-                null,
-            )
+            // Unknown/invalid error body → pass through
+            return first
         }
     }
 

stream-android-core/src/test/java/io/getstream/android/core/internal/http/interceptor/StreamAuthInterceptorTest.kt

@@ -17,7 +17,6 @@ package io.getstream.android.core.internal.http.interceptor
 
 import io.getstream.android.core.api.authentication.StreamTokenManager
 import io.getstream.android.core.api.model.exceptions.StreamEndpointErrorData
-import io.getstream.android.core.api.model.exceptions.StreamEndpointException
 import io.getstream.android.core.api.model.value.StreamToken
 import io.getstream.android.core.api.serialization.StreamJsonSerialization
 import io.mockk.MockKAnnotations
@@ -28,14 +27,15 @@ import io.mockk.impl.annotations.MockK
 import io.mockk.verify
 import java.util.concurrent.TimeUnit
 import kotlin.test.assertEquals
-import kotlin.test.assertFailsWith
 import kotlin.test.assertTrue
 import okhttp3.Interceptor
 import okhttp3.OkHttpClient
 import okhttp3.Request
 import okhttp3.mockwebserver.MockResponse
 import okhttp3.mockwebserver.MockWebServer
 import org.junit.After
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNull
 import org.junit.Before
 import org.junit.Test
 
@@ -187,105 +187,103 @@ class StreamAuthInterceptorTest {
     }
 
     @Test
-    fun `non-token error throws StreamEndpointException without retry`() {
-        val token = streamToken("t1")
+    fun `token error with alreadyRetried header passes through without retry`() {
+        val token = streamToken("stale")
         coEvery { tokenManager.loadIfAbsent() } returns Result.success(token)
 
-        val nonTokenError = tokenErrorData(422)
+        // Proper token error code handled by this interceptor
+        val tokenError = tokenErrorData(40)
         every { json.fromJson(any(), StreamEndpointErrorData::class.java) } returns
-            Result.success(nonTokenError)
+            Result.success(tokenError)
 
         val interceptor = StreamAuthInterceptor(tokenManager, json, authType = "jwt")
         val client = client(interceptor)
 
-        server.enqueue(MockResponse().setResponseCode(422).setBody("""{"error":"unprocessable"}"""))
+        server.enqueue(MockResponse().setResponseCode(401).setBody("""{"error":"token invalid"}"""))
+
+        val url = server.url("/v1/protected")
 
-        val url = server.url("/v1/fail")
-        val ex =
-            assertFailsWith<StreamEndpointException> {
-                client
-                    .newCall(Request.Builder().url(url).build())
-                    .execute()
-                    .use { /* force execution */ }
+        client
+            .newCall(
+                Request.Builder()
+                    .url(url)
+                    .header("x-stream-retried-on-auth", "present") // simulate already retried
+                    .build()
+            )
+            .execute()
+            .use { resp ->
+                assertFalse(resp.isSuccessful) // pass-through, no exception here
+                assertEquals(401, resp.code)
             }
-        assertTrue(ex.message!!.contains("Failed request"))
 
-        // Consume the single (failed) request
         val first = server.takeRequest(2, TimeUnit.SECONDS)
-        kotlin.test.assertNotNull(first, "Expected exactly one request to be sent")
-
-        // Assert no second request (i.e., no retry)
-        val second = server.takeRequest(300, TimeUnit.MILLISECONDS)
-        kotlin.test.assertNull(second, "Interceptor should not retry on non-token errors")
+        kotlin.test.assertNotNull(first)
+        kotlin.test.assertNull(server.takeRequest(300, TimeUnit.MILLISECONDS)) // no second try
 
+        // No refresh/invalidate when header indicates we already retried
         coVerify(exactly = 1) { tokenManager.loadIfAbsent() }
-        io.mockk.verify(exactly = 0) { tokenManager.invalidate() }
+        verify(exactly = 0) { tokenManager.invalidate() }
         coVerify(exactly = 0) { tokenManager.refresh() }
     }
 
+    /** Non-token error codes are NOT handled here; pass response through without retry. */
     @Test
-    fun `unparseable error throws StreamEndpointException`() {
+    fun `non-token error passes through without retry`() {
         val token = streamToken("t1")
         coEvery { tokenManager.loadIfAbsent() } returns Result.success(token)
 
+        // e.g., business error code that is not 40/41/42
+        val nonTokenError = tokenErrorData(13)
         every { json.fromJson(any(), StreamEndpointErrorData::class.java) } returns
-            Result.failure(IllegalStateException("parse error"))
+            Result.success(nonTokenError)
 
         val interceptor = StreamAuthInterceptor(tokenManager, json, authType = "jwt")
         val client = client(interceptor)
 
-        server.enqueue(MockResponse().setResponseCode(500).setBody("not-json"))
+        server.enqueue(MockResponse().setResponseCode(422).setBody("""{"error":"validation"}"""))
 
-        val url = server.url("/v1/error")
-        val ex =
-            assertFailsWith<StreamEndpointException> {
-                client.newCall(Request.Builder().url(url).build()).execute().use { /* consume */ }
-            }
-        assertTrue(ex.message!!.contains("Failed to serialize response error body"))
+        val url = server.url("/v1/endpoint")
+        client.newCall(Request.Builder().url(url).build()).execute().use { resp ->
+            assertFalse(resp.isSuccessful) // still an error; just passed along
+            assertEquals(422, resp.code)
+        }
+
+        // No retry, no token refresh
+        val req = server.takeRequest(2, TimeUnit.SECONDS)!!
+        assertEquals("t1", req.getHeader("Authorization"))
+        kotlin.test.assertNull(server.takeRequest(300, TimeUnit.MILLISECONDS))
 
-        coVerify(exactly = 1) { tokenManager.loadIfAbsent() }
         verify(exactly = 0) { tokenManager.invalidate() }
         coVerify(exactly = 0) { tokenManager.refresh() }
     }
 
+    /** If the error body cannot be parsed into StreamEndpointErrorData, pass through. */
     @Test
-    fun `token error with alreadyRetried header does not retry again`() {
-        val token = streamToken("stale")
+    fun `unparsable error body passes through without retry`() {
+        val token = streamToken("t1")
         coEvery { tokenManager.loadIfAbsent() } returns Result.success(token)
-        every { tokenManager.invalidate() } returns Result.success(Unit)
 
-        val tokenError = tokenErrorData(401)
         every { json.fromJson(any(), StreamEndpointErrorData::class.java) } returns
-            Result.success(tokenError)
+            Result.failure(IllegalStateException("bad json"))
 
         val interceptor = StreamAuthInterceptor(tokenManager, json, authType = "jwt")
         val client = client(interceptor)
 
-        server.enqueue(MockResponse().setResponseCode(401).setBody("""{"error":"token invalid"}"""))
+        server.enqueue(MockResponse().setResponseCode(500).setBody("""<html>oops</html>"""))
 
-        val url = server.url("/v1/protected")
-        val ex =
-            assertFailsWith<StreamEndpointException> {
-                client
-                    .newCall(
-                        Request.Builder()
-                            .url(url)
-                            .header(
-                                "x-stream-retried-on-auth",
-                                "present",
-                            ) // simulate already retried
-                            .build()
-                    )
-                    .execute()
-                    .use { /* consume */ }
-            }
-        assertTrue(ex.message!!.contains("Failed request"))
+        val url = server.url("/v1/boom")
+        client.newCall(Request.Builder().url(url).build()).execute().use { resp ->
+            assertFalse(resp.isSuccessful)
+            assertEquals(500, resp.code)
+        }
 
-        val first = server.takeRequest(2, TimeUnit.SECONDS)
-        kotlin.test.assertNotNull(first)
+        // Consume the single request we expect
+        val first = server.takeRequest(2, TimeUnit.SECONDS)!!
+        assertEquals("t1", first.getHeader("Authorization"))
+
+        // Now verify there's no retry
         kotlin.test.assertNull(server.takeRequest(300, TimeUnit.MILLISECONDS))
 
-        coVerify(exactly = 1) { tokenManager.loadIfAbsent() }
         verify(exactly = 0) { tokenManager.invalidate() }
         coVerify(exactly = 0) { tokenManager.refresh() }
     }