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

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

let events = Events(endpointURL: eventsEndpoint)
let restClient = events.createRestClient(
  publishAuthorizer: APIKeyAuthorizer(apiKey: "apiKey")
)

let restClient = events.createRestClient(
  publishAuthorizer: apiKeyAuthorizer,
  options: .init(
    urlSessionConfiguration: urlSessionConfiguration, // `URLSessionConfiguration`のインスタンス
    logger: AppSyncEventsLogger(), // `EventsLogger`の実装
    interceptor: AppSyncEventsURLRequestInterceptor() // `URLRequestInterceptor`の実装
  )
)

let defaultChannel = "default/channel"
let event = JSONValue(stringLiteral: "123")
do {
  let result = try await restClient.publish(
    channelName: defaultChannel,
    event: event
  )
  print("Publish success with result:\n status: \(result.status) \n, successful events: \(result.successfulEvents) \n, failed events: \(result.failedEvents)")
} catch {
  print("Publish failure with error: \(error)")
}

let defaultChannel = "default/channel"
let eventsList = [
    JSONValue(stringLiteral: "123"),
    JSONValue(booleanLiteral: true),
    JSONValue(floatLiteral: 1.25),
    JSONValue(integerLiteral: 37),
    JSONValue(dictionaryLiteral: ("key", "value"))
]

do {
    let result = try await restClient.publish(
        channelName: defaultChannel,
        events: eventsList
    )
    print("Publish success with result:\n status: \(result.status) \n, successful events: \(result.successfulEvents) \n, failed events: \(result.failedEvents)")
} catch {
  print("Publish failure with error: \(error)")
}

let defaultChannel = "default/channel"
let event = JSONValue(stringLiteral: "123")
let apiKeyAuthorizer = APIKeyAuthorizer(apiKey: "apiKey")
do {
  let result = try await restClient.publish(
    channelName: defaultChannel,
    event: event,
    authorizer: apiKeyAuthorizer
  )
  print("Publish success with result:\n status: \(result.status) \n, successful events: \(result.successfulEvents) \n, failed events: \(result.failedEvents)")
} catch {
  print("Publish failure with error: \(error)")
}

let apiKeyAuthorizer = APIKeyAuthorizer(apiKey: "apiKey")
let webSocketClient = events.createWebSocketClient(
  connectAuthorizer: apiKeyAuthorizer,
  publishAuthorizer: apiKeyAuthorizer,
  subscribeAuthorizer: apiKeyAuthorizer
)

let apiKeyAuthorizer = APIKeyAuthorizer(apiKey: "apiKey")
let webSocketClient = events.createWebSocketClient(
  connectAuthorizer: apiKeyAuthorizer,
  publishAuthorizer: apiKeyAuthorizer,
  subscribeAuthorizer: apiKeyAuthorizer,
  options: .init(
    urlSessionConfiguration: urlSessionConfiguration, // `URLSessionConfiguration`のインスタンス
    logger: AppSyncEventsLogger(), // `EventsLogger`の実装
    interceptor: AppSyncEventsURLRequestInterceptor() // `URLRequestInterceptor`の実装
  ) 
)

let defaultChannel = "default/channel"
let event = JSONValue(stringLiteral: "123")
do {
    let result = try await websocketClient.publish(
        channelName: defaultChannel,
        event: event
    )
    print("Publish success with result:\n status: \(result.status) \n, successful events: \(result.successfulEvents) \n, failed events: \(result.failedEvents)")
} catch {
  print("Publish failure with error: \(error)")
}

let defaultChannel = "default/channel"
let eventsList = [
    JSONValue(stringLiteral: "123"),
    JSONValue(booleanLiteral: true),
    JSONValue(floatLiteral: 1.25),
    JSONValue(integerLiteral: 37),
    JSONValue(dictionaryLiteral: ("key", "value"))
]

do {
    let result = try await websocketClient.publish(
        channelName: defaultChannel,
        events: eventsList
    )
    print("Publish success with result:\n status: \(result.status) \n, successful events: \(result.successfulEvents) \n, failed events: \(result.failedEvents)")
} catch {
  print("Publish failure with error: \(error)")
}

let defaultChannel = "default/channel"
let event = JSONValue(stringLiteral: "123")
let apiKeyAuthorizer = APIKeyAuthorizer(apiKey: "apiKey")
do {
  let result = try await websocketClient.publish(
    channelName: defaultChannel,
    event: event,
    authorizer: apiKeyAuthorizer
  )
  print("Publish success with result:\n status: \(result.status) \n, successful events: \(result.successfulEvents) \n, failed events: \(result.failedEvents)")
} catch {
  print("Publish failure with error: \(error)")
}

let defaultChannel = "default/channel"
let subscription = try websocketClient.subscribe(channelName: defaultChannel)
let task = Task {
    for try await message in subscription {
        print("Subscription received message: \(message))"
    }
}

task.cancel()

let defaultChannel = "default/channel"
let apiKeyAuthorizer = APIKeyAuthorizer(apiKey: "apiKey")
let subscription = try websocketClient.subscribe(
  channelName: defaultChannel,
  authorizer: apiKeyAuthorizer
)
let task = Task {
    for try await message in subscription {
        print("Subscription received message: \(message))"
    }
}

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

import os
import Foundation
import AWSAppSyncEvents

public final class AppSyncEventsLogger: EventsLogger {
    static let lock: NSLocking = NSLock()

    static var _logLevel = LogLevel.error
    
    public init() { }
    
    public var logLevel: LogLevel {
        get {
            AppSyncEventsLogger.lock.lock()
            defer {
                AppSyncEventsLogger.lock.unlock()
            }

            return AppSyncEventsLogger._logLevel
        }
        set {
            AppSyncEventsLogger.lock.lock()
            defer {
                AppSyncEventsLogger.lock.unlock()
            }

            AppSyncEventsLogger._logLevel = newValue
        }
    }

    public func error(_ log: @autoclosure () -> String) {
        os_log("%@", type: .error, log())
    }

    public func error(_ error: @autoclosure () -> Error) {
        os_log("%@", type: .error, error().localizedDescription)
    }

    public func warn(_ log: @autoclosure () -> String) {
        guard logLevel.rawValue >= LogLevel.warn.rawValue else {
            return
        }

        os_log("%@", type: .info, log())
    }

    public func info(_ log: @autoclosure () -> String) {
        guard logLevel.rawValue >= LogLevel.info.rawValue else {
            return
        }

        os_log("%@", type: .info, log())
    }

    public func debug(_ log: @autoclosure () -> String) {
        guard logLevel.rawValue >= LogLevel.debug.rawValue else {
            return
        }

        os_log("%@", type: .debug, log())
    }

    public func verbose(_ log: @autoclosure () -> String) {
        guard logLevel.rawValue >= LogLevel.verbose.rawValue else {
            return
        }

        os_log("%@", type: .debug, log())
    }
}