AWS AppSync Eventsに接続する

import { defineBackend } from '@aws-amplify/backend';
import { auth } from './auth/resource';
Copy highlighted code example
// CDKリソースをインポートします:
import {
  CfnApi,
  CfnChannelNamespace,
  AuthorizationType,
} from 'aws-cdk-lib/aws-appsync';
import { Policy, PolicyStatement } from 'aws-cdk-lib/aws-iam';

const backend = defineBackend({
	auth,
});

Copy highlighted code example
// Event APIリソースの新しいスタックを作成します:
const customResources = backend.createStack('custom-resources');

// スタックに新しいEvent APIを追加します:
const cfnEventAPI = new CfnApi(customResources, 'CfnEventAPI', {
  name: 'my-event-api',
  eventConfig: {
    authProviders: [
      {
        authType: AuthorizationType.USER_POOL,
        cognitoConfig: {
          awsRegion: customResources.region,
          // Event APIをAmplifyによってプロビジョンされたCognito User Poolを使用するように設定します:
          userPoolId: backend.auth.resources.userPool.userPoolId,
        },
      },
    ],
    // Connect、Publish、および Subscribe操作の認可プロバイダーとしてUser Poolを設定します:
    connectionAuthModes: [{ authType: AuthorizationType.USER_POOL }],
    defaultPublishAuthModes: [{ authType: AuthorizationType.USER_POOL }],
    defaultSubscribeAuthModes: [{ authType: AuthorizationType.USER_POOL }],
  },
});

// Event APIのデフォルトネームスペースを作成します:
const namespace = new CfnChannelNamespace(
  customResources,
  'CfnEventAPINamespace',
  {
    apiId: cfnEventAPI.attrApiId,
    name: 'default',
  }
);

// User Pool内の認証されたユーザーロールにポリシーをアタッチしてEvent APIへのアクセスを許可します:
backend.auth.resources.authenticatedUserIamRole.attachInlinePolicy(
  new Policy(customResources, 'AppSyncEventPolicy', {
    statements: [
      new PolicyStatement({
        actions: [
          'appsync:EventConnect',
          'appsync:EventSubscribe',
          'appsync:EventPublish',
        ],
        resources: [`${cfnEventAPI.attrApiArn}/*`, `${cfnEventAPI.attrApiArn}`],
      }),
    ],
  })
);

npx ampx sandbox

val endpoint = "https://abcdefghijklmnopqrstuvwxyz.appsync-api.us-east-1.amazonaws.com/event"
val events = Events(endpoint)

val events: Events // 設定されたEvents
val restClient = events.createRestClient(
  publishAuthorizer = ApiKeyAuthorizer("da2-abcdefghijklmnopqrstuvwxyz")
)

val restClient = events.createRestClient(
    publishAuthorizer = ApiKeyAuthorizer("da2-abcdefghijklmnopqrstuvwxyz"),
    options = Events.Options.Rest(
        loggerProvider = { namespace -> AndroidLogger(namespace, logLevel) },
        okHttpConfigurationProvider = { okHttpBuilder ->
            // EventsRestClientが使用するOkHttp.Builderを更新します
        }
    )
)

val restClient: EventsRestClient // 設定されたEventRestClient
// kotlinx.serialization.json.[JsonElement, JsonPrimitive, JsonArray, JsonObject]
val jsonEvent = JsonObject(mapOf("some" to JsonPrimitive("data")))
val publishResult = restClient.publish(
  channelName = "default/channel", 
  event = jsonEvent
)
when(publishResult) {
  is PublishResult.Response -> {
    val successfulEvents = publishResult.successfulEvents // 成功したイベントをインスペクトします
    val failedEvents = publishResult.failedEvents // 失敗したイベントをインスペクトします
  }
  is PublishResult.Failure -> {
    val error = result.error // パブリッシュ失敗、エラーをインスペクトします
  }
}

val restClient: EventsRestClient // 設定されたEventRestClient
// List of kotlinx.serialization.json.[JsonElement, JsonPrimitive, JsonArray, JsonObject]
val jsonEvents = listOf(
  JsonObject(mapOf("some" to JsonPrimitive("data1"))),
  JsonObject(mapOf("some" to JsonPrimitive("data2")))
)
val publishResult = restClient.publish(
  channelName = "default/channel",
  events = jsonEvents
)
when(publishResult) {
  is PublishResult.Response -> {
    val successfulEvents = publishResult.successfulEvents // 成功したイベントをインスペクトします
    val failedEvents = publishResult.failedEvents // 失敗したイベントをインスペクトします
  }
  is PublishResult.Failure -> {
    val error = result.error // パブリッシュ失敗、エラーをインスペクトします
  }
}

restClient.publish(
  channelName = "default/channel", 
  event = JsonObject(mapOf("some" to JsonPrimitive("data"))),
  authorizer = ApiKeyAuthorizer("da2-abcdefghijklmnopqrstuvwxyz")
)

val events: Events // 設定されたEvents
val apiKeyAuthorizer = ApiKeyAuthorizer("da2-abcdefghijklmnopqrstuvwxyz")
val webSocketClient = events.createWebSocketClient(
  connectAuthorizer = apiKeyAuthorizer, // websocketの接続に使用されます
  subscribeAuthorizer = apiKeyAuthorizer, // サブスクライブコールに使用されます
  publishAuthorizer = apiKeyAuthorizer // パブリッシュコールに使用されます
)

val events: Events // 設定されたEvents
val apiKeyAuthorizer = ApiKeyAuthorizer("da2-abcdefghijklmnopqrstuvwxyz")
val webSocketClient = events.createWebSocketClient(
  connectAuthorizer = apiKeyAuthorizer, // websocketの接続に使用されます
  subscribeAuthorizer = apiKeyAuthorizer, // サブスクライブコールに使用されます
  publishAuthorizer = apiKeyAuthorizer // パブリッシュコールに使用されます,
  options = Events.Options.WebSocket(
    loggerProvider = { namespace -> AndroidLogger(namespace, logLevel) },
    okHttpConfigurationProvider = { okHttpBuilder ->
            // EventsWebSocketClientが使用するOkHttpBuilderを更新します
    }
  )
)

val webSocketClient: EventsWebSocketClient // 設定されたEventsWebSocketClient
// kotlinx.serialization.json.[JsonElement, JsonPrimitive, JsonArray, JsonObject]
val jsonEvent = JsonObject(mapOf("some" to JsonPrimitive("data")))
val publishResult = webSocketClient.publish(
  channelName = "default/channel",
  event = jsonEvent
)
when(publishResult) {
  is PublishResult.Response -> {
    val successfulEvents = publishResult.successfulEvents // 成功したイベントをインスペクトします
    val failedEvents = publishResult.failedEvents // 失敗したイベントをインスペクトします
  }
  is PublishResult.Failure -> {
    val error = result.error // パブリッシュ失敗、エラーをインスペクトします
  }
}

val webSocketClient: EventsWebSocketClient // 設定されたEventsWebSocketClient
// List of kotlinx.serialization.json.[JsonElement, JsonPrimitive, JsonArray, JsonObject]
val jsonEvents = listOf(
  JsonObject(mapOf("some" to JsonPrimitive("data1"))),
  JsonObject(mapOf("some" to JsonPrimitive("data2")))
)
val publishResult = webSocketClient.publish(
  channelName = "default/channel",
  events = jsonEvents
)
when(publishResult) {
  is PublishResult.Response -> {
    val successfulEvents = publishResult.successfulEvents // 成功したイベントをインスペクトします
    val failedEvents = publishResult.failedEvents // 失敗したイベントをインスペクトします
  }
  is PublishResult.Failure -> {
    val error = result.error // パブリッシュ失敗、エラーをインスペクトします
  }
}

val webSocketClient: EventsWebSocketClient // 設定されたEventsWebSocketClient
val jsonEvent = JsonObject(mapOf("some" to JsonPrimitive("data")))
val publishResult = webSocketClient.publish(
  channelName = "default/channel",
  event = jsonEvent,
  authorizer = ApiKeyAuthorizer("da2-abcdefghijklmnopqrstuvwxyz")
)
when(publishResult) {
  is PublishResult.Response -> {
    val successfulEvents = publishResult.successfulEvents // 成功したイベントをインスペクトします
    val failedEvents = publishResult.failedEvents // 失敗したイベントをインスペクトします
  }
  is PublishResult.Failure -> {
    val error = result.error // パブリッシュ失敗、エラーをインスペクトします
  }
}

coroutineScope.launch {
  // subscribeは冷たいFlowを返します。ターミナル演算子が呼び出されると、サブスクリプションが確立されます。
  val subscription: Flow<EventsMessage> = webSocketClient.subscribe("default/channel").onCompletion {
    // サブスクリプションがアンサブスクライブされました
  }.catch { throwable ->
    // サブスクリプションでエラーが発生し、アンサブスクライブされました
    // 原因についてはthrowableを参照してください
  }

  // collectはサブスクリプションを開始します
  subscription.collect { eventsMessage ->
    // JsonElementタイプを返します。
    // Kotlin Serializationを使用して、好みのデータ構造に変換します。
    val jsonData = eventsMessage.data
  }
}

coroutineScope.launch {
  // subscribeは冷たいFlowを返します。ターミナル演算子が呼び出されると、サブスクリプションが確立されます。
  val subscription: Flow<EventsMessage> = webSocketClient.subscribe(
      channelName = "default/channel".
      authorizer = ApiKeyAuthorizer("da2-abcdefghijklmnopqrstuvwxyz")
  ).onCompletion {
    // サブスクリプションがアンサブスクライブされました
  }.catch { throwable ->
    // サブスクリプションでエラーが発生し、アンサブスクライブされました
    // 原因についてはthrowableを参照してください
  }

  // collectはサブスクリプションを開始します
  subscription.collect { eventsMessage ->
    // JsonElementタイプを返します。
    val jsonData = eventsMessage.data
    // Kotlin Serializationを使用して、好みのデータ構造に変換します。
  }
}

val webSocketClient: EventsWebSocketClient // 設定されたEventsWebSocketClient
// 保留中のパブリッシュ操作をWebSocketにポストするのを待つ場合はflushEventsをtrueに設定します
// 保留中のポストを破棄してすぐに切断する場合はflushEventsをfalseに設定します
webSocketClient.disconnect(flushEvents = true) // またはfalseを指定してすぐに切断します

class AndroidLogger(
    private 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)
        }
    }
}