WebSocket으로 객체 집합 구독

TypeScript OSDK의 subscribe는 객체 집합 변경을 WebSocket으로 전달받습니다. 다른 언어에서 연결하거나 프로토콜을 직접 제어하려면 Object Set Watcher 엔드포인트를 사용할 수 있습니다.

연결과 인증

엔드포인트 형식은 다음과 같습니다.

wss://{foundryHost}/api/v2/ontologySubscriptions/ontologies/{ontology}/streamSubscriptions

foundryHost는 https://를 제외한 호스트이고 ontology는 온톨로지 API 이름입니다. WebSocket subprotocol에 Bearer-{token}을 전달합니다. REST의 Authorization: Bearer token과 달리 Bearer 뒤에 공백 대신 하이픈을 사용합니다.

토큰은 앱의 인증 흐름이나 비밀정보 관리 경로에서 제공합니다. URL, 코드와 로그에 비밀값을 넣지 않습니다.

구독 요청

클라이언트와 서버는 JSON 직렬화 메시지를 주고받습니다. 다음 Country, 속성 이름과 UUID는 프로토콜 설명용 값입니다. id는 요청마다 생성하여 응답과 연결합니다.

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "requests": [{
    "objectSet": { "type": "base", "objectType": "Country" },
    "propertySet": ["population", "countryName"],
    "referenceSet": []
  }]
}

필터 집합도 사용할 수 있습니다. 예를 들어 population이 1,000,000 이상인 가상 Country의 정의는 다음과 같습니다.

{
  "type": "filter",
  "objectSet": { "type": "base", "objectType": "Country" },
  "where": { "type": "gte", "field": "population", "value": 1000000 }
}

구독 목록 갱신

새 요청의 requests는 추가분만이 아니라 유지할 전체 구독 목록입니다. 서버는 기존 목록과 비교하여 같은 구독은 유지하고 빠진 것은 닫으며 새 항목은 엽니다.

모든 구독을 해제하려면 새 UUID와 빈 requests를 보냅니다.

{
  "id": "550e8400-e29b-41d4-a716-446655440002",
  "requests": []
}

서버 메시지

subscribeResponses

요청 ID를 돌려주고 각 요청의 성공 또는 오류를 responses에 담습니다. success의 id는 이후 업데이트를 식별하는 구독 ID입니다. 성공 응답의 형식은 다음과 같습니다.

{
  "type": "subscribeResponses",
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "responses": [{ "type": "success", "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890" }]
}

오류 응답은 이름과 진단 인수를 제공합니다.

{
  "type": "subscribeResponses",
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "responses": [{
    "type": "error",
    "errors": [{ "error": "INVALID_OBJECT_TYPE", "args": [{ "name": "objectType", "value": "InvalidType" }] }]
  }]
}

응답의 type이 qos이면 서버 부하가 높은 상태입니다. 지수 백오프에 무작위 지연을 더하여 요청을 다시 보냅니다.

objectSetChanged

구독 ID와 업데이트 목록을 받습니다.

{
  "type": "objectSetChanged",
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "updates": [{
    "type": "object",
    "object": { "__apiName": "Country", "__primaryKey": "US", "countryName": "United States", "population": 331000000 },
    "state": "ADDED_OR_UPDATED"
  }]
}

object.__apiName은 객체 유형 API 이름, __primaryKey는 기본 키입니다. state는 ADDED_OR_UPDATED 또는 REMOVED입니다. REMOVED는 실제 삭제뿐 아니라 필터에서 빠진 경우도 포함합니다. TypeScript 콜백도 같은 REMOVED 상태를 사용합니다.

참조 속성을 구독하면 다음처럼 별도 reference 업데이트를 받을 수 있습니다. 위치 배열은 경도, 위도 순서입니다.

{
  "type": "objectSetChanged",
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "updates": [{
    "type": "reference",
    "objectType": "Aircraft",
    "primaryKey": { "aircraftId": "AC-123" },
    "property": "currentLocation",
    "value": {
      "type": "geotimeSeriesValue",
      "position": { "type": "Point", "coordinates": [-122.4194, 37.7749] },
      "timestamp": "2024-01-15T10:30:00Z"
    }
  }]
}

refreshObjectSet

서버가 증분 업데이트를 제공할 수 없으면 refreshObjectSet과 구독 ID, objectType을 보냅니다. 일반 HTTP API의 POST /api/v2/ontologies/{ontology}/objectSets/loadObjects로 같은 집합을 다시 읽어야 합니다. nextPageToken이 있으면 끝까지 후속 페이지를 가져옵니다.

subscriptionClosed

닫힌 구독의 id와 cause를 받습니다. cause.type이 error이면 SUBSCRIPTION_MEMORY_LIMIT_EXCEEDED 같은 오류와 args가 포함됩니다. 사용자가 요청 목록에서 제거한 경우에는 cause.type이 reason, reason이 USER_CLOSED입니다.

Python 연결 예제

Python 3.10 이상에서 websockets의 asyncio 클라이언트를 사용합니다. FOUNDRY_HOST, FOUNDRY_ONTOLOGY_API_NAME과 FOUNDRY_TOKEN을 외부 실행 환경에 제공해야 합니다. Country와 population, countryName은 자신의 스키마에 맞게 바꿉니다.

이 예제는 객체형 Country 구독의 실제 이벤트를 JSON 줄로 출력하는 수신 프로그램입니다. 시작 및 refresh 시에는 HTTP로 전체 집합을 다시 읽어 snapshot 이벤트를 출력합니다. 출력에 객체 데이터가 포함되므로 실행 환경의 로그와 터미널 접근 권한도 관리해야 합니다.

# /// script
# requires-python = ">=3.10"
# dependencies = ["websockets"]
# ///
import asyncio
import json
import os
import random
import uuid
from urllib.parse import quote
from urllib.request import Request, urlopen
from websockets.asyncio.client import connect
from websockets.exceptions import ConnectionClosed

HOST = os.environ["FOUNDRY_HOST"]
ONTOLOGY = os.environ["FOUNDRY_ONTOLOGY_API_NAME"]
TOKEN = os.environ["FOUNDRY_TOKEN"]
OBJECT_SET = {
    "type": "filter",
    "objectSet": {"type": "base", "objectType": "Country"},
    "where": {"type": "gte", "field": "population", "value": 1000000},
}
PROPERTIES = ["population", "countryName"]


def emit(event: dict) -> None:
    print(json.dumps(event, ensure_ascii=False), flush=True)


def load_snapshot() -> list[dict]:
    url = f"https://{HOST}/api/v2/ontologies/{quote(ONTOLOGY, safe='')}/objectSets/loadObjects"
    body = {"objectSet": OBJECT_SET, "select": PROPERTIES, "pageSize": 1000, "snapshot": True}
    objects: list[dict] = []
    while True:
        request = Request(url, data=json.dumps(body).encode("utf-8"), method="POST", headers={
            "Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json",
        })
        with urlopen(request, timeout=60) as response:
            page = json.load(response)
        objects.extend(page["data"])
        token = page.get("nextPageToken")
        if token is None:
            return objects
        body["pageToken"] = token


async def main() -> None:
    endpoint = f"wss://{HOST}/api/v2/ontologySubscriptions/ontologies/{quote(ONTOLOGY, safe='')}/streamSubscriptions"
    async for websocket in connect(endpoint, subprotocols=[f"Bearer-{TOKEN}"]):
        active: set[str] = set()
        attempt = 0

        async def subscribe() -> None:
            await websocket.send(json.dumps({
                "id": str(uuid.uuid4()),
                "requests": [{"objectSet": OBJECT_SET, "propertySet": PROPERTIES}],
            }))

        try:
            await subscribe()
            async for payload in websocket:
                message = json.loads(payload)
                kind = message["type"]
                if kind == "subscribeResponses":
                    responses = message["responses"]
                    errors = [item for item in responses if item["type"] == "error"]
                    if errors:
                        raise RuntimeError(f"구독 요청 실패: {errors}")
                    if any(item["type"] == "qos" for item in responses):
                        await asyncio.sleep(min(30.0, 2.0 ** min(attempt, 5)) * random.uniform(0.5, 1.5))
                        attempt += 1
                        await subscribe()
                        continue
                    active = {item["id"] for item in responses if item["type"] == "success"}
                    if not active:
                        raise RuntimeError("활성 구독을 받지 못했습니다.")
                    attempt = 0
                    emit({"type": "snapshot", "objects": await asyncio.to_thread(load_snapshot)})
                elif kind == "objectSetChanged":
                    emit(message)
                elif kind == "refreshObjectSet":
                    emit({"type": "stale", "id": message["id"]})
                    emit({"type": "snapshot", "objects": await asyncio.to_thread(load_snapshot)})
                elif kind == "subscriptionClosed":
                    active.discard(message["id"])
                    cause = message["cause"]
                    if cause["type"] == "error":
                        raise RuntimeError(f"구독이 닫혔습니다: {cause}")
                    if not active:
                        return
                else:
                    raise RuntimeError(f"알 수 없는 구독 메시지 유형: {kind}")
        except ConnectionClosed:
            emit({"type": "disconnected"})
            continue


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

connect의 비동기 반복자는 재연결을 관리하며 연결이 다시 열리면 새 요청을 보냅니다. 구독 상태는 서버 재시작을 넘어 유지되지 않으므로 이전 구독 ID를 재사용하지 않습니다. 인증 같은 치명적 연결 실패나 응용 프로토콜 오류는 정상 데이터로 대체하지 않고 전파합니다.

HTTP의 snapshot=True는 페이지를 읽는 동안 같은 조회 스냅샷을 사용합니다. WebSocket 이벤트와 HTTP 스냅샷 사이의 원자적 경계까지 제공하는 것은 아닙니다. 이 프로그램은 이벤트 수신용이며 무손실 변경 로그나 특정 시점의 복제본을 보장하지 않습니다.

인터페이스 속성 매핑

인터페이스 집합의 업데이트도 기반 객체 유형으로 돌아옵니다. __apiName으로 실제 유형을 확인하고 전체 유형 메타데이터를 읽어 인터페이스 필드로 변환합니다.

GET /api/v2/ontologies/{ontology}/objectTypes/{objectType}/fullMetadata?preview=true

이 API는 Preview이며 OAuth 작업 범위 api:ontologies-read가 필요합니다. 응답의 implementsInterfaces2에서 해당 인터페이스 구현을 찾고, 매핑은 구독 수명 동안 캐시하는 것이 권장됩니다.

• properties는 공유 속성 API 이름에서 로컬 속성 API 이름으로의 기존 매핑입니다.
• propertiesV2는 인터페이스 속성 API 이름에서 구현 정의로의 매핑입니다.
• 구현 type이 localPropertyImplementation이면 propertyApiName이 업데이트 객체에서 읽을 로컬 필드입니다.
• structFieldImplementation, structImplementation과 reducedPropertyImplementation은 구조체 필드나 환산 연산이 포함되므로 단순 이름 치환으로 처리하지 않습니다.

가상 매핑 예에서 인터페이스 displayName을 객체의 legalName으로 구현했다면, propertiesV2.displayName의 propertyApiName이 legalName입니다. 업데이트의 legalName 값을 인터페이스 displayName에 연결하는 방향입니다.

제한

조인으로 만든 객체 집합은 완전한 변경 전달을 보장하지 않습니다. 서버가 선택한 부분을 감시하며 그 부분의 변경만 통지할 수 있습니다. 일부 고급 전문 검색 필터도 구독에서 지원하지 않습니다.

구독마다 메모리 한도가 적용되므로 너무 많은 객체를 추적하면 구독이 닫힐 수 있습니다. 클라이언트는 연결 및 구독 종료와 재조회 상태를 처리해야 합니다. TypeScript SDK의 pivotTo 집합은 SDK 구독에서 지원하지 않는 범위입니다.