E2E（エンドツーエンド）コール

ここではiscp-pyを使用したE2Eコールのサンプルを示します。コントローラノードが対象ノードに対して指示を出し、対象ノードは受信完了のリプライを行う簡単なサンプルです。

環境

• python 3.10

事前準備

本サンプルを動作させるための準備をします。 まず、コントローラ用のノードと対象用のノードを作成します。 ノードはintdashのエッジとマッピングされます。intdashAPIや管理コンソールを使用して作成してください。 既に作成したノードがある場合はそのノードIDとシークレットを使用しても問題ありません。 そして環境に合わせて下記に示す値を環境変数としてセットします。

import os
URL_API = os.getenv("URL_API", "https://example.com")
TARGET_NODE_ID = os.getenv("TARGET_NODE_ID", "target-node-id")
TARGET_NODE_SECRET = os.getenv("TARGET_NODE_SECRET", "xxxxxxx")
CONTROLLER_NODE_ID = os.getenv("CONTROLLER_NODE_ID", "controller-node-id")
CONTROLLER_NODE_SECRET = os.getenv("CONTROLLER_NODE_SECRET", "xxxxxxx")

アクセストークンの発行

コントローラ用のアクセストークンと、対象用のアクセストークンをそれぞれ発行します。 ノードのIDとシークレットはOAuth2のクライアントクレデンシャル方式によるトークン発行ができます。

import string

import requests
from oauthlib.oauth2 import BackendApplicationClient
from requests_oauthlib import OAuth2Session

_URL_TOKEN_ENDPOINT = f"{URL_API}/api/auth/oauth2/token"

def token_client_credentials(edge_id: str, secret: str) -> str:
    """
    トークンの払い出しコード
    """
    oauth = OAuth2Session(client=BackendApplicationClient(client_id=edge_id))
    oauth.fetch_token(
        token_url=_URL_TOKEN_ENDPOINT,
        client_id=edge_id,
        client_secret=secret,
        include_client_id=True,
    )
    return oauth.token["access_token"]

target_token=token_client_credentials(TARGET_NODE_ID, TARGET_NODE_SECRET)

controller_token=token_client_credentials(CONTROLLER_NODE_ID, CONTROLLER_NODE_SECRET)

コントローラノードのコード定義

コントローラノードからメッセージを送信するサンプルです。 このサンプルではトランスポートにWebSocketを使用して文字列メッセージを対象ノードに対して送信し、対象ノードからのリプライを待ちます。

import iscp
import asyncio
import urllib

_URL_API = urllib.parse.urlparse(URL_API)
_ENABLE_TLS=_URL_API.scheme == "https"
_ISCP_PORT = _URL_API.port if _URL_API.port else 443 if _URL_API.scheme == "https" else 80
_ADDRESS_WEBSOCKET= f"{_URL_API.hostname}:{_ISCP_PORT}"

connected_target = asyncio.Event()

async def send():
    async with await iscp.Conn.connect(_ADDRESS_WEBSOCKET, iscp.WebSocketConnector(enable_tls=_ENABLE_TLS),
                                       token_source=lambda: controller_token, node_id=CONTROLLER_NODE_ID
                                       ) as conn:
        await connected_target.wait()
        got = await conn.send_call_and_wait_reply_call(
            iscp.UpstreamCall(
                destination_node_id=TARGET_NODE_ID, name="greeting", type="string", payload=b"hello",
            ),
        )
        print(f"Received reply call: {got}")

対象ノードのコード定義

コントローラノードからのコールを受け付け、すぐにリプライをするサンプルです。

import iscp
import json
import dataclasses


async def reply():
    async with await iscp.Conn.connect(_ADDRESS_WEBSOCKET, iscp.WebSocketConnector(enable_tls=_ENABLE_TLS),
                                       token_source=lambda: target_token, node_id=TARGET_NODE_ID
                                       ) as conn:
        connected_target.set()
        call = await conn.calls().__anext__()
        print(f"Received request call: {call}")
        await conn.send_reply_call(
            iscp.UpstreamReplyCall(
                request_call_id=call.call_id,
                destination_node_id=call.source_node_id,
                name="reply_greeting",
                type="string",
                payload=b"world",
            ),
        )

実行

import asyncio
await asyncio.gather(reply(), send())

# 本サンプルはJupyter Notebookから実行しています。
# 通常の利用シーンでは、下記のように非同期処理を実行する必要があることにご注意ください。
# import asyncio

# async def main():
#     await asyncio.gather(reply(), send())

# if __name__ == "__main__":
#     asyncio.run(main())

Received request call: DownstreamCall(call_id='5d9fac92-d3fa-45ed-9009-f2f994a8a6aa', source_node_id='8366f4b1-dc22-4e6d-b22b-dbe2beb92518', name='greeting', type='string', payload=b'hello')
Received reply call: DownstreamReplyCall(call_id='e1d528cf-15b5-4890-92e5-217a75c3a222', request_call_id='5d9fac92-d3fa-45ed-9009-f2f994a8a6aa', source_node_id='49c9458c-7bc9-4ae5-8cff-83a99650fe69', name='reply_greeting', type='string', payload=b'world')

[None, None]