Allow for external event serialization without depending or extending internal external types

Author: aleksandar-apostolov

app/src/main/java/io/getstream/android/core/sample/client/StreamClient.kt

@@ -20,6 +20,7 @@ import io.getstream.android.core.api.authentication.StreamTokenManager
 import io.getstream.android.core.api.authentication.StreamTokenProvider
 import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.log.StreamLoggerProvider
+import io.getstream.android.core.api.model.config.StreamClientSerializationConfig
 import io.getstream.android.core.api.model.value.StreamApiKey
 import io.getstream.android.core.api.model.value.StreamHttpClientInfoHeader
 import io.getstream.android.core.api.model.value.StreamUserId
@@ -28,6 +29,7 @@ import io.getstream.android.core.api.processing.StreamBatcher
 import io.getstream.android.core.api.processing.StreamRetryProcessor
 import io.getstream.android.core.api.processing.StreamSerialProcessingQueue
 import io.getstream.android.core.api.processing.StreamSingleFlightProcessor
+import io.getstream.android.core.api.serialization.StreamProductEventSerialization
 import io.getstream.android.core.api.socket.StreamConnectionIdHolder
 import io.getstream.android.core.api.socket.StreamWebSocketFactory
 import io.getstream.android.core.api.socket.listeners.StreamClientListener
@@ -103,6 +105,11 @@ fun createStreamClient(
         connectionIdHolder = connectionIdHolder,
         socketFactory = socketFactory,
         healthMonitor = healthMonitor,
+        serializationConfig = StreamClientSerializationConfig.default(object :
+            StreamProductEventSerialization<Unit> {
+            override fun serialize(data: Unit): Result<String> = Result.success("")
+            override fun deserialize(raw: String): Result<Unit> = Result.success(Unit)
+        }),
         batcher = batcher,
     )
 }

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

@@ -32,6 +32,7 @@ import io.getstream.android.core.api.processing.StreamRetryProcessor
 import io.getstream.android.core.api.processing.StreamSerialProcessingQueue
 import io.getstream.android.core.api.processing.StreamSingleFlightProcessor
 import io.getstream.android.core.api.serialization.StreamClientEventSerialization
+import io.getstream.android.core.api.serialization.StreamProductEventSerialization
 import io.getstream.android.core.api.socket.StreamConnectionIdHolder
 import io.getstream.android.core.api.socket.StreamWebSocket
 import io.getstream.android.core.api.socket.StreamWebSocketFactory
@@ -40,6 +41,7 @@ import io.getstream.android.core.api.socket.monitor.StreamHealthMonitor
 import io.getstream.android.core.api.subscribe.StreamSubscription
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
 import io.getstream.android.core.internal.client.StreamClientImpl
+import io.getstream.android.core.internal.serialization.StreamCompositeEventSerializationImpl
 import io.getstream.android.core.internal.serialization.StreamCompositeMoshiJsonSerialization
 import io.getstream.android.core.internal.serialization.StreamMoshiJsonSerializationImpl
 import io.getstream.android.core.internal.serialization.moshi.StreamCoreMoshiProvider
@@ -218,8 +220,7 @@ fun StreamClient(
     healthMonitor: StreamHealthMonitor,
     batcher: StreamBatcher<String>,
     // Serialization
-    serializationConfig: StreamClientSerializationConfig =
-        StreamClientSerializationConfig.defaults(),
+    serializationConfig: StreamClientSerializationConfig,
     // Logging
     logProvider: StreamLoggerProvider = StreamLoggerProvider.defaultAndroidLogger(),
 ): StreamClient {
@@ -256,7 +257,6 @@ fun StreamClient(
         serialQueue = serialQueue,
         connectionIdHolder = connectionIdHolder,
         logger = clientLogger,
-        retryProcessor = retryProcessor,
         mutableConnectionState = MutableStateFlow(StreamConnectionState.Idle),
         subscriptionManager = clientSubscriptionManager,
         socketSession =
@@ -271,8 +271,12 @@ fun StreamClient(
                     ),
                 jsonSerialization = compositeSerialization,
                 eventParser =
-                    serializationConfig.eventParser
-                        ?: StreamClientEventSerialization(compositeSerialization),
+                    StreamCompositeEventSerializationImpl(
+                        internal = serializationConfig.eventParser
+                            ?: StreamClientEventSerialization(compositeSerialization),
+                        external = serializationConfig.productEventSerializers,
+                    )
+                    ,
                 healthMonitor = healthMonitor,
                 batcher = batcher,
                 internalSocket = socket,

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

@@ -18,6 +18,7 @@ package io.getstream.android.core.api.model.config
 import io.getstream.android.core.annotations.StreamCoreApi
 import io.getstream.android.core.api.serialization.StreamClientEventSerialization
 import io.getstream.android.core.api.serialization.StreamJsonSerialization
+import io.getstream.android.core.api.serialization.StreamProductEventSerialization
 
 /**
  * Configuration for serialization and deserialization in the Stream client.
@@ -31,31 +32,41 @@ data class StreamClientSerializationConfig
 private constructor(
     val json: StreamJsonSerialization? = null,
     val eventParser: StreamClientEventSerialization? = null,
+    val productEventSerializers: StreamProductEventSerialization<*>,
+    val internalTypes: Set<String> = setOf("connection.ok", "connection.error", "health.check"),
+    val alsoExternal: Set<String> = emptySet(),
 ) {
     companion object {
         /**
          * Creates a default [StreamClientSerializationConfig]. Using the internal implementations.
          *
+         * @param productEvents The product event serializers.
+         * @param alsoExternal The event types to also parse as external.
          * @return A default [StreamClientSerializationConfig].
          */
-        fun defaults() = StreamClientSerializationConfig()
+        fun <T> default(productEvents: StreamProductEventSerialization<T>, alsoExternal: Set<String> = emptySet()) =
+            StreamClientSerializationConfig(productEventSerializers = productEvents, alsoExternal = alsoExternal)
 
         /**
          * Creates a [StreamClientSerializationConfig] with the given JSON serialization.
          *
          * @param serialization The JSON serialization implementation.
+         * @param productEvents The product event serializers.
+         * @param alsoExternal The event types to also parse as external.
          * @return A [StreamClientSerializationConfig] with the given JSON serialization.
          */
-        fun json(serialization: StreamJsonSerialization) =
-            StreamClientSerializationConfig(json = serialization)
+        fun <T> json(serialization: StreamJsonSerialization, productEvents: StreamProductEventSerialization<T>, alsoExternal: Set<String> = emptySet()) =
+            StreamClientSerializationConfig(json = serialization, productEventSerializers = productEvents, alsoExternal = alsoExternal)
 
         /**
          * Creates a [StreamClientSerializationConfig] with the given event parsing.
          *
          * @param serialization The event parsing implementation.
+         * @param productEvents The product event serializers.
+         * @param alsoExternal The event types to also parse as external.
          * @return A [StreamClientSerializationConfig] with the given event parsing.
          */
-        fun event(serialization: StreamClientEventSerialization) =
-            StreamClientSerializationConfig(eventParser = serialization)
+        fun <T> event(serialization: StreamClientEventSerialization, productEvents: StreamProductEventSerialization<T>, alsoExternal: Set<String> = emptySet()) =
+            StreamClientSerializationConfig(eventParser = serialization, productEventSerializers = productEvents, alsoExternal = alsoExternal)
     }
 }

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

@@ -0,0 +1,33 @@
+package io.getstream.android.core.api.serialization
+
+import io.getstream.android.core.annotations.StreamCoreApi
+
+/**
+ * Interface for serializing and deserializing product event objects.
+ *
+ * All operations return a [Result] so callers can safely propagate failures (e.g. malformed input,
+ * I/O errors, schema mismatches) without throwing.
+ *
+ * @param T The product event type handled by this interface.
+ */
+@StreamCoreApi
+interface StreamProductEventSerialization<T> {
+
+    /**
+     * Encodes a product event into a [String] suitable for transport or storage.
+     *
+     * @param data The event to serialize.
+     * @return `Result.success(String)` when encoding succeeds, or `Result.failure(Throwable)` when
+     *   the process fails.
+     */
+    fun serialize(data: T): Result<String>
+
+    /**
+     * Decodes a product event from a [String] representation.
+     *
+     * @param raw The string to deserialize.
+     * @return `Result.success(T)` when decoding succeeds, or `Result.failure(Throwable)` when
+     *   the process fails.
+     */
+    fun deserialize(raw: String): Result<T>
+}

stream-android-core/src/main/java/io/getstream/android/core/api/socket/listeners/StreamClientListener.kt

@@ -39,7 +39,7 @@ interface StreamClientListener {
      *
      * @param event The event received from the WebSocket.
      */
-    fun onEvent(event: StreamClientWsEvent) {}
+    fun onEvent(event: Any) {}
 
     /**
      * Called when an error occurs on the client.

stream-android-core/src/main/java/io/getstream/android/core/internal/client/StreamClientImpl.kt

@@ -21,9 +21,7 @@ import io.getstream.android.core.api.log.StreamLogger
 import io.getstream.android.core.api.model.StreamTypedKey.Companion.randomExecutionKey
 import io.getstream.android.core.api.model.connection.StreamConnectedUser
 import io.getstream.android.core.api.model.connection.StreamConnectionState
-import io.getstream.android.core.api.model.event.StreamClientWsEvent
 import io.getstream.android.core.api.model.value.StreamUserId
-import io.getstream.android.core.api.processing.StreamRetryProcessor
 import io.getstream.android.core.api.processing.StreamSerialProcessingQueue
 import io.getstream.android.core.api.processing.StreamSingleFlightProcessor
 import io.getstream.android.core.api.socket.StreamConnectionIdHolder
@@ -39,14 +37,13 @@ import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.update
 
-internal class StreamClientImpl(
+internal class StreamClientImpl<T>(
     private val userId: StreamUserId,
     private val tokenManager: StreamTokenManager,
     private val singleFlight: StreamSingleFlightProcessor,
     private val serialQueue: StreamSerialProcessingQueue,
-    private val retryProcessor: StreamRetryProcessor,
     private val connectionIdHolder: StreamConnectionIdHolder,
-    private val socketSession: StreamSocketSession,
+    private val socketSession: StreamSocketSession<T>,
     private val mutableConnectionState: MutableStateFlow<StreamConnectionState>,
     private val logger: StreamLogger,
     private val subscriptionManager: StreamSubscriptionManager<StreamClientListener>,
@@ -77,13 +74,14 @@ internal class StreamClientImpl(
                     socketSession
                         .subscribe(
                             object : StreamClientListener {
+
                                 override fun onState(state: StreamConnectionState) {
                                     logger.v { "[client#onState]: $state" }
                                     mutableConnectionState.update(state)
                                     subscriptionManager.forEach { it.onState(state) }
                                 }
 
-                                override fun onEvent(event: StreamClientWsEvent) {
+                                override fun onEvent(event: Any) {
                                     logger.v { "[client#onEvent]: $event" }
                                     subscriptionManager.forEach { it.onEvent(event) }
                                 }

stream-android-core/src/main/java/io/getstream/android/core/internal/serialization/StreamCompositeEventSerializationImpl.kt

@@ -0,0 +1,132 @@
+package io.getstream.android.core.internal.serialization
+
+import com.squareup.moshi.JsonReader
+import io.getstream.android.core.api.model.event.StreamClientWsEvent
+import io.getstream.android.core.api.serialization.StreamClientEventSerialization
+import io.getstream.android.core.api.serialization.StreamProductEventSerialization
+import io.getstream.android.core.api.utils.runCatchingCancellable
+import okio.Buffer
+
+/**
+ * Represents a composite event that can be either a [StreamClientWsEvent] or a product-specific event.
+ *
+ * @param T The type of the product-specific event.
+ */
+internal class StreamCompositeSerializationEvent<T> private constructor(
+    val core: StreamClientWsEvent? = null,
+    val product: T? = null,
+) {
+    companion object {
+        /**
+         * Creates a new [StreamCompositeSerializationEvent] with the given [event].
+         *
+         * @param event The event to wrap.
+         * @return A new [StreamCompositeSerializationEvent] with the given [event].
+         */
+        fun <T> internal(event: StreamClientWsEvent) = StreamCompositeSerializationEvent<T>(core = event)
+
+        /**
+         * Creates a new [StreamCompositeSerializationEvent] with the given [event].
+         *
+         * @param event The event to wrap.
+         * @return A new [StreamCompositeSerializationEvent] with the given [event].
+         */
+        fun <T> external(event: T) = StreamCompositeSerializationEvent(product = event)
+
+        /**
+         * Creates a new [StreamCompositeSerializationEvent] with the given [event].
+         *
+         * @param event The event to wrap.
+         * @return A new [StreamCompositeSerializationEvent] with the given [event].
+         */
+        fun <T> both(event: StreamClientWsEvent, product: T) =
+            StreamCompositeSerializationEvent(core = event, product = product)
+    }
+}
+
+/**
+ * Serializes and deserializes [StreamCompositeSerializationEvent] objects.
+ *
+ * @param T The type of the product-specific event.
+ */
+internal class StreamCompositeEventSerializationImpl<T>(
+    private val internal: StreamClientEventSerialization,
+    private val external: StreamProductEventSerialization<T>,
+    private val internalTypes: Set<String> = setOf("connection.ok", "connection.error", "health.check"),
+    private val alsoExternal: Set<String> = emptySet(),
+) {
+    /**
+     * Serializes the given [data] into a [String].
+     *
+     * @param data The data to serialize.
+     * @return `Result.success(String)` when encoding succeeds, or `Result.failure(Throwable)` when
+     *   the process fails.
+     */
+    fun serialize(data: StreamCompositeSerializationEvent<T>): Result<String> {
+        data.core?.let { return internal.serialize(it) }
+        data.product?.let { return external.serialize(it) }
+        return Result.failure(NullPointerException())
+    }
+
+    /**
+     * Deserializes the given [raw] string into a [StreamCompositeSerializationEvent].
+     *
+     * @param raw The string to deserialize.
+     * @return `Result.success(StreamCompositeSerializationEvent)` when decoding succeeds, or
+     *   `Result.failure(Throwable)` when the process fails.
+     */
+    fun deserialize(raw: String): Result<StreamCompositeSerializationEvent<T>> =
+        runCatchingCancellable {
+            val type = peekType(raw)
+            return try {
+                when (type) {
+                    null -> {
+                        val ext = external
+                        ext.deserialize(raw).map { StreamCompositeSerializationEvent.external(it) }
+                    }
+                    in alsoExternal -> {
+                        val coreSer = internal
+                        val extSer  = external
+                        val core = coreSer.deserialize(raw).getOrThrow()
+                        val prod = extSer.deserialize(raw).getOrThrow()
+                        Result.success(StreamCompositeSerializationEvent.both(core, prod))
+                    }
+                    in internalTypes -> {
+                        val coreSer = internal
+                        coreSer.deserialize(raw).map { StreamCompositeSerializationEvent.internal(it) }
+                    }
+                    else -> {
+                        val ext = external
+                        ext.deserialize(raw).map { StreamCompositeSerializationEvent.external(it) }
+                    }
+                }
+            } catch (e: Throwable) {
+                Result.failure(e)
+            }
+        }
+
+    private fun peekType(raw: String): String? {
+        val reader = JsonReader.of(Buffer().writeUtf8(raw))
+        reader.isLenient = true
+        return try {
+            if (reader.peek() != JsonReader.Token.BEGIN_OBJECT) return null
+            reader.beginObject()
+            var result: String? = null
+            while (reader.hasNext()) {
+                val name = reader.nextName()
+                if (name == "type" && reader.peek() == JsonReader.Token.STRING) {
+                    result = reader.nextString()
+                    // consume the rest to keep reader state valid
+                    while (reader.hasNext()) reader.skipValue()
+                    break
+                } else {
+                    reader.skipValue()
+                }
+            }
+            if (reader.peek() == JsonReader.Token.END_OBJECT) reader.endObject()
+            result
+        } catch (_: Throwable) {
+            null
+        }
+    }
+}