feat(data): Events Lib consistent exception handling (#3028)

Author: tylerjroach

appsync/aws-appsync-events/src/main/java/com/amplifyframework/aws/appsync/events/Events.kt

@@ -19,6 +19,7 @@ import com.amplifyframework.aws.appsync.core.LoggerProvider
 import com.amplifyframework.aws.appsync.events.data.ChannelAuthorizers
 import com.amplifyframework.aws.appsync.events.data.EventsException
 import com.amplifyframework.aws.appsync.events.data.PublishResult
+import com.amplifyframework.aws.appsync.events.data.toEventsException
 import kotlinx.coroutines.coroutineScope
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonElement
@@ -92,7 +93,11 @@ class Events @VisibleForTesting internal constructor(
         event: JsonElement,
         authorizer: AppSyncAuthorizer = this.defaultChannelAuthorizers.publishAuthorizer
     ): PublishResult {
-        return httpClient.post(channelName, authorizer, event)
+        return try {
+            httpClient.post(channelName, authorizer, event)
+        } catch (exception: Exception) {
+            throw exception.toEventsException()
+        }
     }
 
     /**
@@ -109,7 +114,11 @@ class Events @VisibleForTesting internal constructor(
         events: List<JsonElement>,
         authorizer: AppSyncAuthorizer = this.defaultChannelAuthorizers.publishAuthorizer
     ): PublishResult {
-        return httpClient.post(channelName, authorizer, events)
+        return try {
+            httpClient.post(channelName, authorizer, events)
+        } catch (exception: Exception) {
+            throw exception.toEventsException()
+        }
     }
 
     /**
@@ -129,8 +138,6 @@ class Events @VisibleForTesting internal constructor(
      *
      * @param flushEvents set to true (default) to allow all pending publish calls to succeed before disconnecting.
      * Setting to false will immediately disconnect, cancelling any in-progress or queued event publishes.
-     * @param authorizers for the channel to use for subscriptions and publishes.
-     * @return a channel to manage subscriptions and publishes.
      */
     suspend fun disconnect(flushEvents: Boolean = true): Unit = coroutineScope {
         eventsWebSocketProvider.existingWebSocket?.disconnect(flushEvents)

appsync/aws-appsync-events/src/main/java/com/amplifyframework/aws/appsync/events/EventsChannel.kt

@@ -22,6 +22,7 @@ import com.amplifyframework.aws.appsync.events.data.EventsMessage
 import com.amplifyframework.aws.appsync.events.data.PublishResult
 import com.amplifyframework.aws.appsync.events.data.UserClosedConnectionException
 import com.amplifyframework.aws.appsync.events.data.WebSocketMessage
+import com.amplifyframework.aws.appsync.events.data.toEventsException
 import java.util.UUID
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.async
@@ -83,12 +84,16 @@ class EventsChannel internal constructor(
      * @param authorizer for the publish call. If not provided, the EventChannel publish authorizer will be used.
      * @return result of publish.
      */
-    @Throws(Exception::class)
+    @Throws(EventsException::class)
     suspend fun publish(
         event: JsonElement,
         authorizer: AppSyncAuthorizer = this.authorizers.publishAuthorizer
     ): PublishResult {
-        return publish(listOf(event), authorizer)
+        return try {
+            publishToWebSocket(listOf(event), authorizer)
+        } catch (exception: Exception) {
+            throw exception.toEventsException()
+        }
     }
 
     /**
@@ -102,6 +107,17 @@ class EventsChannel internal constructor(
     suspend fun publish(
         events: List<JsonElement>,
         authorizer: AppSyncAuthorizer = this.authorizers.publishAuthorizer
+    ): PublishResult {
+        return try {
+            publishToWebSocket(events, authorizer)
+        } catch (exception: Exception) {
+            throw exception.toEventsException()
+        }
+    }
+
+    private suspend fun publishToWebSocket(
+        events: List<JsonElement>,
+        authorizer: AppSyncAuthorizer
     ): PublishResult = coroutineScope {
         val publishId = UUID.randomUUID().toString()
         val publishMessage = WebSocketMessage.Send.Publish(

appsync/aws-appsync-events/src/main/java/com/amplifyframework/aws/appsync/events/EventsWebSocket.kt

@@ -21,6 +21,7 @@ import com.amplifyframework.aws.appsync.core.LoggerProvider
 import com.amplifyframework.aws.appsync.events.data.ConnectException
 import com.amplifyframework.aws.appsync.events.data.EventsException
 import com.amplifyframework.aws.appsync.events.data.WebSocketMessage
+import com.amplifyframework.aws.appsync.events.data.toEventsException
 import com.amplifyframework.aws.appsync.events.utils.ConnectionTimeoutTimer
 import com.amplifyframework.aws.appsync.events.utils.HeaderKeys
 import com.amplifyframework.aws.appsync.events.utils.HeaderValues

appsync/aws-appsync-events/src/main/java/com/amplifyframework/aws/appsync/events/RestClient.kt

@@ -17,7 +17,10 @@ package com.amplifyframework.aws.appsync.events
 
 import com.amplifyframework.aws.appsync.core.AppSyncAuthorizer
 import com.amplifyframework.aws.appsync.core.AppSyncRequest
+import com.amplifyframework.aws.appsync.events.data.EventsErrors
+import com.amplifyframework.aws.appsync.events.data.EventsException
 import com.amplifyframework.aws.appsync.events.data.PublishResult
+import com.amplifyframework.aws.appsync.events.data.toEventsException
 import com.amplifyframework.aws.appsync.events.utils.HeaderKeys
 import com.amplifyframework.aws.appsync.events.utils.HeaderValues
 import kotlinx.serialization.json.Json
@@ -74,15 +77,17 @@ internal class RestClient(
             }
         }.build()
 
-        try {
-            val result = okHttpClient.newCall(authRequest).execute()
-            return if (result.isSuccessful) {
-                json.decodeFromString<PublishResult>(result.body.string())
-            } else {
-                TODO("Convert to proper exception type")
+        val result = okHttpClient.newCall(authRequest).execute()
+        val body = result.body.string()
+        return if (result.isSuccessful) {
+            json.decodeFromString<PublishResult>(body)
+        } else {
+            throw try {
+                val errors = json.decodeFromString<EventsErrors>(body)
+                errors.toEventsException("Failed to post event(s)")
+            } catch (e: Exception) {
+                EventsException.unknown("Failed to post event(s)", e)
             }
-        } catch (e: Exception) {
-            TODO("Convert to proper exception type")
         }
     }
 }

appsync/aws-appsync-events/src/main/java/com/amplifyframework/aws/appsync/events/data/EventsError.kt

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2025 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License").
+ *  You may not use this file except in compliance with the License.
+ *  A copy of the License is located at
+ *
+ *   http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file 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.amplifyframework.aws.appsync.events.data
+
+import com.amplifyframework.aws.appsync.events.data.WebSocketMessage.Received
+import kotlinx.serialization.Serializable
+
+@Serializable
+internal data class EventsErrors(val errors: List<EventsError>) : Received()
+
+internal fun EventsErrors.toEventsException(fallbackMessage: String? = null): EventsException {
+    return errors.firstOrNull()?.toEventsException(fallbackMessage) ?: EventsException.unknown(fallbackMessage)
+}
+
+@Serializable
+internal data class EventsError(val errorType: String, val message: String? = null)
+
+// fallback message is only used if WebSocketError didn't provide a message
+internal fun EventsError.toEventsException(fallbackMessage: String? = null): EventsException {
+    val message = this.message ?: fallbackMessage
+    return when (errorType) {
+        "UnauthorizedException" -> UnauthorizedException(message)
+        "BadRequestException" -> BadRequestException(message)
+        "MaxSubscriptionsReachedError" -> MaxSubscriptionsReachedException(message)
+        "LimitExceededError" -> RateLimitExceededException(message)
+        "ResourceNotFound" -> ResourceNotFoundException(message)
+        "UnsupportedOperation" -> UnsupportedOperationException(message)
+        "InvalidInputError" -> InvalidInputException(message)
+        else -> EventsException(message = "$errorType: $message")
+    }
+}

appsync/aws-appsync-events/src/main/java/com/amplifyframework/aws/appsync/events/data/EventsException.kt

@@ -14,6 +14,8 @@
  */
 package com.amplifyframework.aws.appsync.events.data
 
+import java.net.UnknownHostException
+
 /**
  * Base class for exceptions thrown in Events library
  *
@@ -28,15 +30,27 @@ open class EventsException internal constructor(
 ) : Exception(message, cause) {
 
     internal companion object {
-        internal fun unknown(message: String? = null): EventsException {
+        internal fun unknown(
+            message: String? = null,
+            cause: Throwable? = null
+        ): EventsException {
             return EventsException(
                 message = message ?: "An unknown error occurred",
+                cause = cause,
                 recoverySuggestion = "This is not expected to occur. Contact AWS"
             )
         }
     }
 }
 
+fun Exception.toEventsException(): EventsException {
+    return when (this) {
+        is EventsException -> this
+        is UnknownHostException -> NetworkException(throwable = this)
+        else -> EventsException.unknown(cause = this)
+    }
+}
+
 /**
  * Thrown when failing to connect to Events WebSocket.
  */
@@ -66,43 +80,64 @@ class ConnectionClosedException internal constructor(cause: Throwable? = null) :
 /**
  * Thrown when rate limit is exceeded.
  */
-internal class RateLimitExceededException internal constructor() : EventsException(
-    message = "Rate limit exceeded",
+class RateLimitExceededException internal constructor(message: String?) : EventsException(
+    message = message ?: "Rate limit exceeded",
     recoverySuggestion = "Try again later"
 )
 
 /**
  * Thrown when operation is unsupported.
  */
-internal class UnsupportedOperationException internal constructor() : EventsException(
-    message = "WebSocket did not understand the operation",
+class UnsupportedOperationException internal constructor(message: String?) : EventsException(
+    message = message ?: "WebSocket did not understand the operation",
     recoverySuggestion = "This is not expected to occur. Contact AWS"
 )
 
 /**
  * Thrown when resource is not found.
  */
-internal class ResourceNotFoundException internal constructor() : EventsException(
-    message = "Resource not found",
-    recoverySuggestion = "Check Event configuration values and try again"
+class ResourceNotFoundException internal constructor(message: String?) : EventsException(
+    message = message ?: "Namespace not found",
+    recoverySuggestion = "Check resource values and try again"
 )
 
 /**
  * Thrown when hitting max subscription limit.
  */
-class MaxSubscriptionsReachedException internal constructor(throwable: Throwable) : EventsException(
-    message = "Max number of subscriptions reached",
+class MaxSubscriptionsReachedException internal constructor(message: String?) : EventsException(
+    message = message ?: "Max number of subscriptions reached",
     recoverySuggestion = "Unsubscribe from existing channels before attempting to subscribe."
 )
 
 /**
  * Thrown when attempting to send too many events or invalid request.
  */
-class BadRequestException internal constructor() : EventsException(
-    message = "Input exceeded 5 event limit",
-    recoverySuggestion = "Submit 5 events or less."
+class BadRequestException internal constructor(message: String?) : EventsException(
+    message = message ?: "An unknown error occurred"
+)
+
+/**
+ * Thrown when attempting to send too many events or invalid request over websocket.
+ */
+class InvalidInputException internal constructor(message: String?) : EventsException(
+    message = message ?: "An unknown error occurred"
 )
 
+/**
+ * Thrown when we detect a failure in the network.
+ * See the cause for the underlying error.
+ */
+class NetworkException internal constructor(throwable: Throwable) : EventsException(
+    message = "Network error",
+    cause = throwable,
+    recoverySuggestion = "Check your internet connection and try again. See the cause for more details."
+)
+
+/**
+ * An internal exception that is not provided to the customer.
+ * We use this exception so that we can differentiate between a connection being closed unexpectedly or by the user.
+ * If the connection is closed by the user, we catch this exception and don't propagate it to the customer.
+ */
 internal class UserClosedConnectionException internal constructor() : EventsException(
     message = "The websocket connection was closed normally"
 )

appsync/aws-appsync-events/src/main/java/com/amplifyframework/aws/appsync/events/data/WebSocketMessage.kt

@@ -83,7 +83,7 @@ internal sealed class WebSocketMessage {
         internal data object KeepAlive : Received()
 
         @Serializable @SerialName("connection_error")
-        internal data class ConnectionError(val errors: List<WebSocketError>) : Received()
+        internal data class ConnectionError(val errors: List<EventsError>) : Received()
 
         @Serializable
         internal sealed class Subscription : Received() {
@@ -101,13 +101,13 @@ internal sealed class WebSocketMessage {
             @Serializable @SerialName("subscribe_error")
             internal data class SubscribeError(
                 override val id: String,
-                override val errors: List<WebSocketError>
+                override val errors: List<EventsError>
             ) : Subscription(), ErrorContainer
 
             @Serializable @SerialName("unsubscribe_error")
             internal data class UnsubscribeError(
                 override val id: String,
-                override val errors: List<WebSocketError>
+                override val errors: List<EventsError>
             ) : Subscription(), ErrorContainer
         }
 
@@ -121,13 +121,13 @@ internal sealed class WebSocketMessage {
         @Serializable @SerialName("publish_error")
         data class PublishError(
             override val id: String? = null,
-            override val errors: List<WebSocketError>
+            override val errors: List<EventsError>
         ) : Received(), ErrorContainer
 
         @Serializable @SerialName("error")
         data class Error(
             override val id: String? = null,
-            override val errors: List<WebSocketError>
+            override val errors: List<EventsError>
         ) : Received(), ErrorContainer
     }
 
@@ -136,19 +136,6 @@ internal sealed class WebSocketMessage {
     // All errors contain an id and errors list
     internal interface ErrorContainer {
         val id: String?
-        val errors: List<WebSocketError>
-    }
-}
-
-@Serializable
-data class WebSocketError(val errorType: String, val message: String? = null) {
-
-    // fallback message is only used if WebSocketError didn't provide a message
-    fun toEventsException(fallbackMessage: String? = null): EventsException {
-        val message = this.message ?: fallbackMessage
-        return when (errorType) {
-            "UnauthorizedException" -> UnauthorizedException(message)
-            else -> EventsException(message = "$errorType: $message")
-        }
+        val errors: List<EventsError>
     }
 }