feat(data): Complete events class

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

@@ -56,6 +56,7 @@ class Events @VisibleForTesting internal constructor(
     }
     private val endpoints = EventsEndpoints(endpoint)
     private val httpClient = RestClient(endpoints.restEndpoint, okHttpClient, json)
+    private val eventsWebSocket = EventsWebSocket(endpoints, connectAuthorizer, okHttpClient, json)
 
     /**
      * Publish a single event to a channel.
@@ -98,13 +99,10 @@ class Events @VisibleForTesting internal constructor(
      * @param authorizers for the channel to use for subscriptions and publishes.
      * @return a channel to manage subscriptions and publishes.
      */
-    @Throws(EventsException::class)
     fun channel(
         channelName: String,
         authorizers: ChannelAuthorizers = this.defaultChannelAuthorizers,
-    ): EventsChannel {
-        TODO("Need to implement")
-    }
+    ) = EventsChannel(channelName, authorizers, endpoints, eventsWebSocket)
 
     /**
      * Method to disconnect from all channels.
@@ -114,8 +112,7 @@ class Events @VisibleForTesting internal constructor(
      * @param authorizers for the channel to use for subscriptions and publishes.
      * @return a channel to manage subscriptions and publishes.
      */
-    @Throws(EventsException::class)
     suspend fun disconnect(flushEvents: Boolean = true) {
-        TODO("Need to implement")
+        eventsWebSocket.disconnect(flushEvents)
     }
 }

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

@@ -30,7 +30,9 @@ import kotlinx.serialization.json.JsonElement
  */
 class EventsChannel internal constructor(
     val name: String,
-    val authorizers: ChannelAuthorizers
+    val authorizers: ChannelAuthorizers,
+    private val endpoints: EventsEndpoints,
+    private val eventsWebSocket: EventsWebSocket
 ) {
 
     /**

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

@@ -0,0 +1,33 @@
+/*
+ * 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
+
+import com.amplifyframework.aws.appsync.core.AppSyncAuthorizer
+import kotlinx.serialization.json.Json
+import okhttp3.OkHttpClient
+import okhttp3.WebSocketListener
+
+internal class EventsWebSocket(
+    private val eventsEndpoints: EventsEndpoints,
+    private val authorizer: AppSyncAuthorizer,
+    private val okHttpClient: OkHttpClient,
+    private val json: Json
+) : WebSocketListener() {
+
+    suspend fun disconnect(flushEvents: Boolean = true) {
+        TODO("Not yet implemented")
+    }
+}

appsync/aws-appsync-events/src/test/java/com/amplifyframework/aws/appsync/events/EventsTest.kt

@@ -47,14 +47,15 @@ class EventsTest {
         HeaderKeys.CONTENT_TYPE to HeaderValues.CONTENT_TYPE_APPLICATION_JSON,
         HeaderKeys.ACCEPT to HeaderValues.ACCEPT_APPLICATION_JSON,
     )
+    private val expectedChannelAuthorizers = ChannelAuthorizers(
+        subscribeAuthorizer = ApiKeyAuthorizer("123"),
+        publishAuthorizer = TestAuthorizer()
+    )
     private val interceptor = ConvertToMockRequestInterceptor(mockWebServer.url("/event"))
     private val events = Events(
         endpoint = expectedEndpoint,
         connectAuthorizer = ApiKeyAuthorizer("abc"),
-        defaultChannelAuthorizers = ChannelAuthorizers(
-            subscribeAuthorizer = ApiKeyAuthorizer("123"),
-            publishAuthorizer = TestAuthorizer()
-        ),
+        defaultChannelAuthorizers = expectedChannelAuthorizers,
         okHttpClient = OkHttpClient.Builder()
             .addInterceptor(interceptor)
             .build()
@@ -304,6 +305,46 @@ class EventsTest {
             successfulEvents.size shouldBe 1
         }
     }
+
+    @Test
+    fun `test channel creation with default authorizers`() = runTest {
+        // GIVEN
+        val expectedChannel = "default/testChannel"
+
+        // WHEN
+        val channel = events.channel(expectedChannel)
+
+        // THEN
+        channel.name shouldBe expectedChannel
+        channel.authorizers shouldBe expectedChannelAuthorizers
+    }
+
+    @Test
+    fun `test channel creation with override authorizers`() = runTest {
+        // GIVEN
+        val expectedChannel = "default/testChannel"
+        val overrideChannelAuthorizers = ChannelAuthorizers(
+            subscribeAuthorizer = ApiKeyAuthorizer("override"),
+            publishAuthorizer = TestAuthorizer("override")
+        )
+
+        // WHEN
+        val channel = events.channel(expectedChannel, overrideChannelAuthorizers)
+
+        // THEN
+        channel.name shouldBe expectedChannel
+        channel.authorizers shouldBe overrideChannelAuthorizers
+    }
+
+    @Test
+    fun `test disconnect with flushEvents`() {
+        // TODO: Write test once disconnect implemented within websocket
+    }
+
+    @Test
+    fun `test disconnect without flushEvents`() {
+        // TODO: Write test once disconnect implemented within websocket
+    }
 }
 
 private class ConvertToMockRequestInterceptor(private val mockUrl: HttpUrl) : Interceptor {