スケーラブルなメッセージング – Momento Topics

Introduction

先日MomentoのPub/Sub機能についての記事を書きましたが、
３月末に「Momento Topics」という名称で正式にリリースされました。

Momento Topicsはイベント駆動型アーキテクチャを構築するための
シンプルなメッセージングサービスです。
Momentoのキャッシュ機能と同じく、設定不要、自動スケーラブル、
金額は従量制の親切設計になっています。

今回は正式リリースされたMomento Topicsをjavascript用SDKからつかってみます。

Environment

今回試した環境は以下のとおりです。

• MacBook Pro (13-inch, M1, 2020)
• OS : MacOS 13.0.1
• Momento CLI : v0.30.0
• Node : v18.15.0

DevIOのMomento関連記事はここにあるので、
これらもご確認ください。

Setup

例によってまずはMomentoセットアップです。
このあたりを参考に、
MomentoのCLIインストールと認証トークンを取得しましょう。

認証トークンを取得したらコンソールでCLI用・プログラム用に
トークンをセットしておきます。

# CLI用
% momento configure
Please paste your Momento auth token.  
(If you do not have an auth token, use `momento account` to generate one.)
Windows users: if CTRL-V does not work, try right-click or SHIFT-INSERT to paste.

Token [****]:<認証トークン>

　・・・

#プログラム用
% export MOMENTO_AUTH_TOKEN = <取得した認証トークン>

サンプル用プロジェクトの作成後、必要ライブラリをインストールしておきましょう。

% mkdir topics && cd topics
% npm install -g typescript
% npm install -g -D ts-node
% npm install @gomomento/sdk

↓の main.tsファイルを記述します。

// main.ts
function hello(name: string) {
  console.log(`Hello, ${name}!`);
}

hello('World');

ts-nodeで実行できればOK。
これで準備完了です。

% ts-node main.ts
Hello, World!

Try

ではここを参考に、typescriptで
Momento Topicsの　PublisherとSubscriberをそれぞれ実装してみます。
まだTopic用に使うキャッシュがなければ、CLIでキャッシュを作成しておきましょう。

% momento cache create --name <キャッシュ名>

Publisherの実装

ではTypeScriptで Publisherを実装します。
pub.tsという名前で下記のように実装します。
cacheNameには作成したキャッシュ名を設定します。

import {
  TopicClient,
  TopicPublish,
  Configurations,
  CredentialProvider,
} from '@gomomento/sdk';

const cacheName = "<キャッシュ名>";
const topicName = "my_topic";
const message = "hello";

async function main() {

  const momento = new TopicClient({
    configuration: Configurations.Laptop.v1(),
    credentialProvider: CredentialProvider.fromEnvironmentVariable({
      environmentVariableName: 'MOMENTO_AUTH_TOKEN',
    }),
  });

 

  console.log(
    `Publish cacheName=${cacheName}, topicName=${topicName}, message=${message}`
  );
  const publishResponse = await momento.publish(cacheName, topicName, message);
  if (publishResponse instanceof TopicPublish.Success) {
    console.log('success published');
  } else {
    console.log(`Error : ${publishResponse.toString()}`);
  }
}

main()
  .then(() => {
    console.log('success!!');
  })
  .catch((e: Error) => {
    console.error(`Error: ${e.message}`);
    throw e;
  });

TopicClientをインスタンス化後、
キャッシュ名とトピック名を指定してpublishするだけでOKです、

次はSubscriberの実装。 sub.tsという名前で実装します。

import {
  TopicClient,
  TopicItem,
  TopicSubscribe,
  Configurations,
  CredentialProvider,
} from '@gomomento/sdk';

const cacheName = "<キャッシュ名>";
const topicName = "my_topic";

async function main() {
  const momento = new TopicClient({
    configuration: Configurations.Laptop.v1(),
    credentialProvider: CredentialProvider.fromEnvironmentVariable({
      environmentVariableName: 'MOMENTO_AUTH_TOKEN',
    }),
  });

  console.log(`Subscribe cacheName=${cacheName}, topicName=${topicName}`);

  const response = await momento.subscribe(cacheName, topicName, {
    onItem: handleItem,
    onError: handleError,
  });

  if (response instanceof TopicSubscribe.Subscription) {
    console.log('Subscribed');
  } else if (response instanceof TopicSubscribe.Error) {
    console.log(`Error : ${response.toString()}`);
    return;
  } else {
    console.log(
      `response: ${response.toString()}`
    );
    return;
  }

  const sleep = (seconds: number) =>
    new Promise(r => setTimeout(r, seconds * 1000));

  await sleep(120);

  if (response instanceof TopicSubscribe.Subscription) {
    console.log(
      'Unsubscribing from topic subscription. Restart the example to subscribe again.'
    );
    response.unsubscribe();
  }
}

function handleItem(item: TopicItem) {
  console.log('Item received from topic subscription; %s', item);
}

function handleError(
  error: TopicSubscribe.Error,
  subscription: TopicSubscribe.Subscription
) {
  console.log(`Error : ${error.toString()}`);
}

main()
  .then(() => {
    console.log('success');
  })
  .catch((e: Error) => {
    console.error(`Error: ${e.message}`);
    throw e;
  });

こちらも実装はシンプルです。
TopicClientをインスタンス化したらsubscribeをよぶだけ。
実際に実行してみます。
まずsub.tsを実行してSubscriberを起動。

% ts-node sub.ts
[2023-04-18T08:38:29.814Z] INFO (Momento: TopicClient): Creating Momento CacheClient
Subscribing to cacheName=my_cache, topicName=my_topic
Subscribed to topic

別ウィンドウでPublisherを実行します。

% ts-node pub.ts
[2023-04-18T08:39:10.208Z] INFO (Momento: TopicClient): Creating Momento CacheClient
Publishing cacheName=my_cache, topicName=my_topic, value=hello
success published

Subscriberがメッセージを受信してますね。

Item received from topic subscription; TopicItem: hello

Summary

TypeScriptでMomento Topicsを使ってみました。
今回は基本機能を試しただけですが、
今後便利な機能がどんどん実装される予定です。

最新情報については、↓で紹介しているユーザーグループのイベントや、
定期的に開催しているMomentoセミナーでお知らせするので、
ぜひご参加ください。

なお、Momentoについてのお問い合わせはこちらです。
こちらもお気軽にお問い合わせください。

Announcement about Momento

先日、Momentoのユーザーグループが発足し、
4/24(月)に第１回イベントを開催することになりました。
ゲストスピーカーを招いてのセッション、Momentoコアメンバーのご紹介、
Momentoの最新情報発表などを予定しているので、ぜひご参加ください。

References

• Momentoがさらにパワーアップ：Topics を紹介
• Serverless CacheサービスのMomentoがPub/Sub機能を実装します