Momento Topics（pub/sub）API を使用する

Momento Topics は、分散型アプリケーションの各部分間でリアルタイム通信を可能にするメッセージングパターンです。トピックの値をパブリッシュ(プロデュース)し、トピックからサブスクライブ(コンシューム)することを可能にします。このページでは、Momento Topics で操作するための Momento API メソッドについて詳しく説明します。

詳細はMomento Topicsをご覧ください。

Topics のメソッド

Subscribe

このメソッドでは、ステートフルな接続を用いて新しい値を受け取るためにトピックをサブスクライブします。

名前	型	説明
cacheName	String	トピックが存在するキャッシュの名前
topicName	String	サブスクライブするトピックの名前

これが サンプルコードです。

Coming soon.

メソッドのレスポンスオブジェクト

• Success - サブスクリプションオブジェクトを返します。
• Error

具体的な情報についてはレスポンスオブジェクトをご覧ください。

返されたサブスクリプションオブジェクトをforループに置くと、新しい値がトピックに公開される時にコードにイベントが送信されます。

JavaScript

const result = await topicClient.subscribe(cacheName, 'test-topic', {
  onError: () => {
    return;
  },
  onItem: (item: TopicItem) => {
    console.log(`Received an item on subscription for 'test-topic': ${item.value().toString()}`);
    return;
  },
});
if (result instanceof TopicSubscribe.Subscription) {
  console.log("Successfully subscribed to topic 'test-topic'");

  console.log("Publishing a value to the topic 'test-topic'");
  // Publish a value
  await topicClient.publish(cacheName, 'test-topic', 'test-value');

  console.log('Waiting for the published value to be received.');
  await new Promise(resolve => setTimeout(resolve, 1000));

  // Need to close the stream before the example ends or else the example will hang.
  result.unsubscribe();
} else if (result instanceof TopicSubscribe.Error) {
  throw new Error(
    `An error occurred while attempting to subscribe to the topic 'test-topic' in cache '${cacheName}': ${result.errorCode()}: ${result.toString()}`
  );
}

Python

response = await topic_client.subscribe("cache", "my_topic")
match response:
    case TopicSubscribe.Error() as error:
        print(f"Error subscribing to topic: {error.message}")
    case TopicSubscribe.SubscriptionAsync() as subscription:
        await topic_client.publish("cache", "my_topic", "my_value")
        async for item in subscription:
            match item:
                case TopicSubscriptionItem.Text():
                    print(f"Received message as string: {item.value}")
                    return
                case TopicSubscriptionItem.Binary():
                    print(f"Received message as bytes: {item.value!r}")
                    return
                case TopicSubscriptionItem.Error():
                    print(f"Error with received message: {item.inner_exception.message}")
                    return

Kotlin

when (val response = topicClient.subscribe("test-cache", "test-topic")) {
    is TopicSubscribeResponse.Subscription -> coroutineScope {
        launch {
            withTimeoutOrNull(2000) {
                response.collect { item ->
                    when (item) {
                        is TopicMessage.Text -> println("Received text message: ${item.value}")
                        is TopicMessage.Binary -> println("Received binary message: ${item.value}")
                        is TopicMessage.Error -> throw RuntimeException(
                            "An error occurred reading messages from topic 'test-topic': ${item.errorCode}", item
                        )
                    }
                }
            }
        }
    }

    is TopicSubscribeResponse.Error -> throw RuntimeException(
        "An error occurred while attempting to subscribe to topic 'test-topic': ${response.errorCode}", response
    )
}

Go

// Instantiate subscriber
sub, subErr := client.Subscribe(ctx, &momento.TopicSubscribeRequest{
	CacheName: "test-cache",
	TopicName: "test-topic",
})
if subErr != nil {
	panic(subErr)
}

time.Sleep(time.Second)
_, pubErr := client.Publish(ctx, &momento.TopicPublishRequest{
	CacheName: "test-cache",
	TopicName: "test-topic",
	Value:     momento.String("test-message"),
})
if pubErr != nil {
	panic(pubErr)
}
time.Sleep(time.Second)

item, err := sub.Item(ctx)
if err != nil {
	panic(err)
}
switch msg := item.(type) {
case momento.String:
	fmt.Printf("received message as string: '%v'\n", msg)
case momento.Bytes:
	fmt.Printf("received message as bytes: '%v'\n", msg)
}

C#

var produceCancellation = new CancellationTokenSource();
produceCancellation.CancelAfter(2000);

var subscribeResponse = await topicClient.SubscribeAsync("test-cache", "test-topic");
switch (subscribeResponse)
{
    case TopicSubscribeResponse.Subscription subscription:
        var cancellableSubscription = subscription.WithCancellation(produceCancellation.Token);

        await Task.Delay(1_000);
        await topicClient.PublishAsync("test-cache", "test-topic", "test-topic-value");
        await Task.Delay(1_000);

        await foreach (var message in cancellableSubscription)
        {
            switch (message)
            {
                case TopicMessage.Binary:
                    Console.WriteLine("Received unexpected binary message from topic.");
                    break;
                case TopicMessage.Text text:
                    Console.WriteLine($"Received string message from topic: {text.Value}");
                    break;
                case TopicMessage.Error error:
                    throw new Exception($"An error occurred while receiving topic message: {error.ErrorCode}: {error}");
                default:
                    throw new Exception("Bad message received");
            }
        }
        subscription.Dispose();
        break;
    case TopicSubscribeResponse.Error error:
        throw new Exception($"An error occurred subscribing to a topic: {error.ErrorCode}: {error}");
}

Swift

let subscribeResponse = await topicClient.subscribe(cacheName: cacheName, topicName: "topic")

#if swift(>=5.9)
let subscription = switch subscribeResponse {
    case .error(let err): fatalError("Error subscribing to topic: \(err)")
    case .subscription(let sub): sub
}
#else 
let subscription: TopicSubscription
switch subscribeResponse {
    case .error(let err):
        fatalError("Error subscribing to topic: \(err)")
    case .subscription(let sub):
        subscription = sub
}
#endif

// unsubscribe in 5 seconds
Task {
    try await Task.sleep(nanoseconds: 5_000_000_000)
    subscription.unsubscribe()
}

// loop over messages as they are received
for try await item in subscription.stream {
    var value: String = ""
    switch item {
    case .itemText(let textItem):
        value = textItem.value
        print("Subscriber recieved text message: \(value)")
    case .itemBinary(let binaryItem):
        value = String(decoding: binaryItem.value, as: UTF8.self)
        print("Subscriber recieved binary message: \(value)")
    case .error(let err):
        print("Subscriber received error: \(err)")
    }
}

Dart

final subscription = await topicClient.subscribe("test-cache", "test-topic");
final messageStream = switch (subscription) {
  TopicSubscription() => subscription.stream,
  TopicSubscribeError() => throw Exception(
      "Subscribe error: ${subscription.errorCode} ${subscription.message}"),
};

// cancel subscription 5 seconds from now
Timer(const Duration(seconds: 5), () {
  print("Cancelling subscription!");
  subscription.unsubscribe();
});

try {
  await for (final msg in messageStream) {
    switch (msg) {
      case TopicSubscriptionItemBinary():
        print("Binary value: ${msg.value}");
      case TopicSubscriptionItemText():
        print("String value: ${msg.value}");
    }
  }
} catch (e) {
  print("Runtime type: ${e.runtimeType}");
  print("Error with await for loop: $e");
}

Publish

メッセージをトピックにパブリッシュします。

名前	型	説明
cacheName	String	トピックが存在するキャッシュの名前
topicName	String	値をパブリッシュするトピック名
value	String / bytes	トピックにパブリッシュする値

こちらがサンプルコードです。

Coming soon.

メソッドのレスポンスオブジェクト

• Success
• Error

具体的な情報についてはレスポンスオブジェクトをご覧ください。

JavaScript

const result = await topicClient.publish(cacheName, 'test-topic', 'test-topic-value');
if (result instanceof TopicPublish.Success) {
  console.log("Value published to topic 'test-topic'");
} else if (result instanceof TopicPublish.Error) {
  throw new Error(
    `An error occurred while attempting to publish to the topic 'test-topic' in cache '${cacheName}': ${result.errorCode()}: ${result.toString()}`
  );
}

Python

response = await topic_client.publish("cache", "my_topic", "my_value")
match response:
    case TopicPublish.Success():
        print("Successfully published a message")
    case TopicPublish.Error() as error:
        print(f"Error publishing a message: {error.message}")

Kotlin

when (val response = topicClient.publish("test-cache", "test-topic", "test-message")) {
    is TopicPublishResponse.Success -> println("Message published successfully")
    is TopicPublishResponse.Error -> throw RuntimeException(
        "An error occurred while attempting to publish message to topic 'test-topic': ${response.errorCode}",
        response
    )
}

Go

_, err := client.Publish(ctx, &momento.TopicPublishRequest{
	CacheName: "test-cache",
	TopicName: "test-topic",
	Value:     momento.String("test-message"),
})
if err != nil {
	panic(err)
}

C#

var publishResponse =
     await topicClient.PublishAsync("test-cache", "test-topic", "test-topic-value");
switch (publishResponse)
{
    case TopicPublishResponse.Success:
        Console.WriteLine("Successfully published message to 'test-topic'");
        break;
    case TopicPublishResponse.Error error:
        throw new Exception($"An error occurred while publishing topic message: {error.ErrorCode}: {error}");
}

Swift

let result = await topicClient.publish(
    cacheName: cacheName,
    topicName: "topic",
    value: "value"
)
switch result {
case .success(_):
    print("Successfully published message!")
case .error(let err):
    print("Unable to publish message: \(err)")
    exit(1)
}

Dart

final result = await topicClient.publish("cache", "topic", "hello message!");
switch (result) {
  case TopicPublishSuccess():
    print("Successful publish!");
  case TopicPublishError():
    print("Publish error: ${result.errorCode} ${result.message}");
}

TopicClient

ほとんどの Momento Cache API コールでは CacheClient を使用していますが、Topics については TopicClient オブジェクトを使用します。

JavaScript

new TopicClient({
  configuration: TopicConfigurations.Default.latest(),
  credentialProvider: CredentialProvider.fromEnvironmentVariable({
    environmentVariableName: 'MOMENTO_API_KEY',
  }),
});

Python

TopicClientAsync(
    TopicConfigurations.Default.latest(), CredentialProvider.from_environment_variable("MOMENTO_API_KEY")
)

Kotlin

TopicClient(
    CredentialProvider.fromEnvVar("MOMENTO_API_KEY"), TopicConfigurations.Laptop.latest
).use { topicClient ->
    //...
}

Go

credProvider, err := auth.NewEnvMomentoTokenProvider("MOMENTO_API_KEY")
if err != nil {
	panic(err)
}

topicClient, err := momento.NewTopicClient(
	config.TopicsDefault(),
	credProvider,
)
if err != nil {
	panic(err)
}

C#

new TopicClient(
    TopicConfigurations.Laptop.latest(),
    new EnvMomentoTokenProvider("MOMENTO_API_KEY")
);

Swift

do {
    let credentialProvider = try CredentialProvider.fromEnvironmentVariable(envVariableName: "MOMENTO_API_KEY")
    let topicClient = TopicClient(
        configuration: TopicClientConfigurations.iOS.latest(),
        credentialProvider: credentialProvider
    )
} catch {
    print("Unable to create topic client: \(error)")
    exit(1)
}

Dart

try {
  final topicClient = TopicClient(
      CredentialProvider.fromEnvironmentVariable("MOMENTO_API_KEY"),
      TopicClientConfigurations.latest());
} catch (e) {
  print("Unable to create topic client: $e");
  exit(1);
}

Example apps using Momento Topics APIs

A growing list of example apps using the Momento Topics.

Momento Topicsを使用したアプリの例が続々と増えています。

• サーバーレスで作成されたアイテムをパブリッシュするマイクロサービス このマイクロサービスはTypeScriptで書かれ、API Gateway、Lambda関数、Momento Topicsを使ってAWS上で実行される。(API Gateway上で適切なセキュリティが設定されていれば）他のサービスでも利用することができ、様々なトピックにメッセージを発行して他のアプリケーションから購読させることができます。この API に topicName と topicValue を渡すと、このサービスはその値をトピックにパブリッシュします。