Skip to content

feat(data): Events WebSocket Connection Logic + Subscribe #3018

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Apr 11, 2025
Merged

feat(data): Events WebSocket Connection Logic + Subscribe #3018

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b3c52fd
Websocket subscription handling
tylerjroach Apr 11, 2025
2d1bca3
Logic allowing singular websocket and shared usage across channels
tylerjroach Apr 11, 2025
06e236f
async/await should suspend, not block, regardless of dispatcher
tylerjroach Apr 11, 2025
722c510
lint
tylerjroach Apr 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
222 changes: 222 additions & 0 deletions appsync/aws-appsync-core/src/main/java/com/amplifyframework/aws/appsync/core/util/Logger.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,222 @@
/*
* 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.core.util

import java.util.function.Supplier

/**
* A component which can emit logs.
*/
interface Logger {
/**
* Gets the minimum log-level, below which no logs are emitted.
* This value is assigned when obtaining and instance of the logger.
* @return The minimum permissible LogLevel for which logs will be emitted
*/
val thresholdLevel: LogLevel

/**
* Gets the namespace of the logger.
* @return namespace for logger
*/
val namespace: String
Copy link
Member

@mattcreaser mattcreaser Apr 14, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

namespace appears to be unused, is it supposed to be used as the tag of the logcat message?

The way I'd kind of expected this to work is for there to be an interface like:

interface LoggerFactory {
     fun logger(namespace: String): Logger
}

And then a particular events class (say EventsWebSocket) would do something like val logger = loggerFactory.logger("EventsWebSocket") so that it wouldn't have to prepend a TAG to every message.

Copy link
Member Author

@tylerjroach tylerjroach Apr 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently, I have it set up like this:

private val events = Events(
    endpoint = endpoint,
    connectAuthorizer = apiKeyAuthorizer,
    defaultChannelAuthorizers = ChannelAuthorizers(
        subscribeAuthorizer = apiKeyAuthorizer,
        publishAuthorizer = apiKeyAuthorizer
    ),
    options = Events.Options(AndroidLogger("Events", LogLevel.DEBUG))
)

/**
 * Implementation of Logger interface that logs to Android's LogCat
 */
class AndroidLogger(
    override val namespace: String,
    override val thresholdLevel: LogLevel
) : Logger {

    override fun error(message: String) {
        if (!thresholdLevel.above(LogLevel.ERROR)) {
            Log.e(namespace, message)
        }
    }

    override fun error(message: String, error: Throwable?) {
        if (!thresholdLevel.above(LogLevel.ERROR)) {
            Log.e(namespace, message, error)
        }
    }

    override fun warn(message: String) {
        if (!thresholdLevel.above(LogLevel.WARN)) {
            Log.w(namespace, message)
        }
    }

    override fun warn(message: String, issue: Throwable?) {
        if (!thresholdLevel.above(LogLevel.WARN)) {
            Log.w(namespace, message, issue)
        }
    }

    override fun info(message: String) {
        if (!thresholdLevel.above(LogLevel.INFO)) {
            Log.i(namespace, message)
        }
    }

    override fun debug(message: String) {
        if (!thresholdLevel.above(LogLevel.DEBUG)) {
            Log.d(namespace, message)
        }
    }

    override fun verbose(message: String) {
        if (!thresholdLevel.above(LogLevel.VERBOSE)) {
            Log.v(namespace, message)
        }
    }
}

This was also on customer code side but it would be reasonable for us to include the implementation for AndroidLogger. This is all the same as how our current Amplify Logger works.

Copy link
Member Author

@tylerjroach tylerjroach Apr 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed after discussion.


/**
* Logs a message at the [LogLevel.ERROR] level.
* @param message An error message
*/
fun error(message: String)

/**
* Logs a message at the [LogLevel.ERROR] level. The supplier is only invoked if the log level threshold
* is at ERROR or below.
* @param messageSupplier A function that returns an error message
*/
fun error(messageSupplier: Supplier<String>) {
if (!thresholdLevel.above(LogLevel.ERROR)) {
error(messageSupplier.get())
}
}

/**
* Logs a message and thrown error at [LogLevel.ERROR] level.
* @param message An error message
* @param error A thrown error
*/
fun error(message: String, error: Throwable?)

/**
* Logs a message and thrown error at [LogLevel.ERROR] level. The supplier is only invoked if the log level
* threshold is at ERROR or below.
* @param error A thrown error
* @param messageSupplier A function that returns an error message
*/
fun error(error: Throwable?, messageSupplier: Supplier<String>) {
if (!thresholdLevel.above(LogLevel.ERROR)) {
error(messageSupplier.get(), error)
}
}

/**
* Log a message at the [LogLevel.WARN] level.
* @param message A warning message
*/
fun warn(message: String)

/**
* Log a message at the [LogLevel.WARN] level. The supplier is only invoked if the log level threshold
* is at WARN or below.
* @param messageSupplier A function that returns a warning message
*/
fun warn(messageSupplier: Supplier<String>) {
if (!thresholdLevel.above(LogLevel.WARN)) {
warn(messageSupplier.get())
}
}

/**
* Log a message and a throwable issue at the [LogLevel.WARN] level.
* @param message A warning message
* @param issue An issue that caused this warning
*/
fun warn(message: String, issue: Throwable?)

/**
* Log a message and a throwable issue at the [LogLevel.WARN] level. The supplier is only invoked if the
* log level threshold is at WARN or below.
* @param issue An issue that caused this warning
* @param messageSupplier A function that returns a warning message
*/
fun warn(issue: Throwable?, messageSupplier: Supplier<String>) {
if (!thresholdLevel.above(LogLevel.WARN)) {
warn(messageSupplier.get(), issue)
}
}

/**
* Logs a message at [LogLevel.INFO] level.
* @param message An informational message
*/
fun info(message: String)

/**
* Logs a message at [LogLevel.INFO] level. The supplier is only invoked if the log level threshold
* is at INFO or below.
* @param messageSupplier A function that returns an info message
*/
fun info(messageSupplier: Supplier<String>) {
if (!thresholdLevel.above(LogLevel.INFO)) {
info(messageSupplier.get())
}
}

/**
* Logs a message at the [LogLevel.DEBUG] level.
* @param message A debugging message.
*/
fun debug(message: String)

/**
* Logs a message at the [LogLevel.DEBUG] level. The supplier is only invoked if the log level threshold
* is at DEBUG or below.
* @param messageSupplier A function that returns a debugging message
*/
fun debug(messageSupplier: Supplier<String>) {
if (!thresholdLevel.above(LogLevel.DEBUG)) {
debug(messageSupplier.get())
}
}

/**
* Logs a message at the [LogLevel.VERBOSE] level.
* @param message A verbose message
*/
fun verbose(message: String)

/**
* Logs a message at the [LogLevel.VERBOSE] level. The supplier is only invoked if the log level threshold
* is at VERBOSE.
* @param messageSupplier A function that returns a verbose message
*/
fun verbose(messageSupplier: Supplier<String>) {
if (!thresholdLevel.above(LogLevel.VERBOSE)) {
verbose(messageSupplier.get())
}
}
}

/**
* An enumeration of the different levels of logging.
* The levels are progressive, with lower-value items being lower priority
* than higher-value items. For example, INFO is lower priority than WARNING
* or ERROR.
*/
enum class LogLevel {
/**
* Verbose logs are used to study the behavior of particular components/flows
* within a system, by developers. Verbose logs are not suitable for emission
* in production, as they may contain sensitive information, and/or be emitted
* so frequently that performance is impacted.
*/
VERBOSE,

/**
* Debug logs are useful during development to understand the behavior of the system.
* These logs may contain information that is inappropriate for emission in a production
* environment.
*/
DEBUG,

/**
* Informational logs may be emitted in production code, and provide
* terse information about the general operation and flow of a piece of software.
*/
INFO,

/**
* Warning logs indicate potential issues while running a piece of software.
* For example, a network connection might retry, before succeeding with success -
* the system is functioning without error, but not optimally.
*/
WARN,

/**
* Errors should be logged when the system is not operating as expected.
* For example, there might be no internet connection available to the system.
* The application probably shouldn't need to crash, but anything that needs the
* Internet will error out. Errors may logically be recoverable or fatal, but are
* not distinguished here, and are logged at this same error level.
*/
ERROR,

/**
* A log level above all other log levels. This log level may be used as a threshold
* value, to prevent any logs from being emitted.
*/
NONE;

/**
* Checks if a log level is above the current level.
* For example, NONE is above ERROR, but ERROR is not above ERROR,
* and WARN is not above ERROR.
* @param threshold A threshold level to consider for evaluation
* @return true if the current level is above the threshold level
* @throws NullPointerException if threshold is null
*/
fun above(threshold: LogLevel): Boolean {
return this.ordinal > threshold.ordinal
}
}
32 changes: 26 additions & 6 deletions appsync/aws-appsync-events/src/main/java/com/amplifyframework/aws/appsync/events/Events.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,11 @@
package com.amplifyframework.aws.appsync.events

import com.amplifyframework.aws.appsync.core.AppSyncAuthorizer
import com.amplifyframework.aws.appsync.core.util.Logger
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 kotlinx.coroutines.coroutineScope
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.JsonElement
import okhttp3.OkHttpClient
Expand All @@ -34,9 +36,14 @@ class Events @VisibleForTesting internal constructor(
val endpoint: String,
val connectAuthorizer: AppSyncAuthorizer,
val defaultChannelAuthorizers: ChannelAuthorizers,
options: Options,
okHttpClient: OkHttpClient
) {

data class Options(
val logger: Logger? = null
)

/**
* The main class for interacting with AWS AppSync Events
*
Expand All @@ -47,16 +54,29 @@ class Events @VisibleForTesting internal constructor(
constructor(
endpoint: String,
connectAuthorizer: AppSyncAuthorizer,
defaultChannelAuthorizers: ChannelAuthorizers
) : this(endpoint, connectAuthorizer, defaultChannelAuthorizers, OkHttpClient.Builder().build())
defaultChannelAuthorizers: ChannelAuthorizers,
options: Options = Options()
) : this(
endpoint,
connectAuthorizer,
defaultChannelAuthorizers,
options,
OkHttpClient.Builder().build()
)

private val json = Json {
encodeDefaults = true
ignoreUnknownKeys = true
}
private val endpoints = EventsEndpoints(endpoint)
private val httpClient = RestClient(endpoints.restEndpoint, okHttpClient, json)
private val eventsWebSocket = EventsWebSocket(endpoints, connectAuthorizer, okHttpClient, json)
private val eventsWebSocketProvider = EventsWebSocketProvider(
endpoints,
connectAuthorizer,
okHttpClient,
json,
options.logger
)

/**
* Publish a single event to a channel.
Expand Down Expand Up @@ -102,7 +122,7 @@ class Events @VisibleForTesting internal constructor(
fun channel(
channelName: String,
authorizers: ChannelAuthorizers = this.defaultChannelAuthorizers,
) = EventsChannel(channelName, authorizers, endpoints, eventsWebSocket)
) = EventsChannel(channelName, authorizers, endpoints, eventsWebSocketProvider)

/**
* Method to disconnect from all channels.
Expand All @@ -112,7 +132,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.
*/
suspend fun disconnect(flushEvents: Boolean = true) {
eventsWebSocket.disconnect(flushEvents)
suspend fun disconnect(flushEvents: Boolean = true): Unit = coroutineScope {
eventsWebSocketProvider.getExistingWebSocket()?.disconnect(flushEvents)
}
}
Loading