WebSocket API

WebSocket API (WebSocket)

Pulsar WebSocket API는 공식 클라이언트 라이브러리가 없는 언어에서도 Pulsar와 상호작용할 수 있는 간단한 방법을 제공해요. 어떤 WebSocket 클라이언트 라이브러리와도 함께 사용할 수 있어요. 이 문서는 WebSocket 서비스 실행 방법과 프로듀서·컨슈머·리더 세 엔드포인트의 사용법을 정리했어요.

출처: 문서

본문

Pulsar WebSocket API는 공식 클라이언트 라이브러리가 없는 언어를 사용해 Pulsar와 상호작용하는 간단한 방법을 제공해요. 어떤 WebSocket 클라이언트 라이브러리와도 Pulsar WebSocket API를 사용할 수 있어요. 자세한 내용은 Python과 Node.js 예시를 참고하세요.

WebSocket을 통해 메시지를 게시하고 소비할 수 있으며, 클라이언트 기능 매트릭스 페이지에서 사용할 수 있는 기능들을 이용할 수 있어요.

WebSocket 서비스 실행 (Run the WebSocket service)

로컬 개발에 권장하는 Pulsar의 standalone 변형은 이미 WebSocket 서비스가 활성화되어 있어요.

비-standalone 모드에서는 WebSocket 서비스를 배포하는 두 가지 방법이 있어요.

  • Pulsar 브로커에 내장(embedded)
  • 별도 컴포넌트로

Pulsar 브로커에 내장 (Embedded with a Pulsar broker)

이 모드에서 WebSocket 서비스는 브로커에서 이미 실행 중인 같은 HTTP 서비스 안에서 실행돼요. 이 모드를 활성화하려면 설치본의 conf/broker.conf 구성 파일에서 webSocketServiceEnabled 파라미터를 설정하세요.

webSocketServiceEnabled=true

별도 컴포넌트로 (As a separate component)

이 모드에서 WebSocket 서비스는 Pulsar 브로커에서 별도 서비스로 실행돼요. 이 모드의 구성은 conf/websocket.conf 구성 파일에서 처리돼요. 다음 파라미터는 최소한 설정해야 해요.

  • configurationMetadataStoreUrl
  • webServicePort
  • clusterName

예시:

configurationMetadataStoreUrl=zk1:2181,zk2:2181,zk3:2181
webServicePort=8080
clusterName=my-cluster

보안 설정 (Security settings)

WebSocket 서비스에 TLS 암호화를 활성화하려면 conf/broker.conf 파일에서 다음 파라미터를 구성해요.

tlsEnabled=true
tlsAllowInsecureConnection=false
tlsCertificateFilePath=/path/to/client-websocket.cert.pem
tlsKeyFilePath=/path/to/client-websocket.key-pk8.pem
tlsTrustCertsFilePath=/path/to/ca.cert.pem

WebSocket 서비스에 저장 데이터 암호화(encryption at rest)를 활성화하려면 WebSocket용 CryptoKeyReader를 만들 CryptoKeyReaderFactory 팩토리 클래스를 classpath에 추가해 프로듀서/컨슈머의 암호화 키를 로드하세요.

cryptoKeyReaderFactoryClassName=org.apache.pulsar.MyCryptoKeyReaderFactoryClassImpl

브로커 시작 (Start the broker)

구성이 설정되면 pulsar-daemon 도구를 사용해 서비스를 시작할 수 있어요.

bin/pulsar-daemon start websocket

릴리스 노트 (Release notes)

Pulsar WebSocket API의 변경 로그는 릴리스 노트를 참고하세요.

API 참조 (API Reference)

Pulsar의 WebSocket API는 메시지를 생산(produce), 소비(consume), 읽기(read) 위한 세 가지 엔드포인트를 제공해요.

WebSocket API를 통한 모든 교환은 JSON을 사용해요.

인증 (Authentication)

보안상의 이유로 쿼리 파라미터가 아니라 헤더로 토큰을 전달하는 것을 권장해요.

쿼리 파라미터 token을 사용해 인증 토큰을 전달해요.

ws://broker-service-url:8080/path?token=token

프로듀서 엔드포인트 (Producer endpoint)

프로듀서 엔드포인트는 URL에 테넌트, 네임스페이스, 토픽을 지정해야 해요.

ws://broker-service-url:8080/ws/v2/producer/persistent/:tenant/:namespace/:topic

쿼리 파라미터 (Query param)

키 타입 필수? 설명
sendTimeoutMillis long 아니요 Send 타임아웃 (기본: 30초)
batchingEnabled boolean 아니요 메시지 배칭 활성화 (기본: false)
batchingMaxMessages int 아니요 배치에 허용되는 최대 메시지 수 (기본: 1000)
maxPendingMessages int 아니요 메시지를 보관하는 내부 큐의 최대 크기 설정 (기본: 1000)
batchingMaxPublishDelay long 아니요 메시지가 배치되는 시간 (기본: 10ms)
messageRoutingMode string 아니요 파티셔닝된 프로듀서의 메시지 라우팅 모드: SinglePartition, RoundRobinPartition
compressionType string 아니요 압축 타입: LZ4, ZLIB
producerName string 아니요 프로듀서의 이름을 지정해요. Pulsar는 같은 이름의 프로듀서 하나만 토픽에 게시할 수 있도록 강제해요.
initialSequenceId long 아니요 프로듀서가 게시하는 메시지의 시퀀스 ID 기준선을 설정해요.
hashingScheme string 아니요 파티셔닝된 토픽에 게시할 때 사용할 해시 함수: JavaStringHash, Murmur3_32Hash
token string 아니요 인증 토큰. 브라우저 javascript 클라이언트에 사용돼요.
encryptionKeys string 아니요 게시된 메시지를 암호화할 암호화 키. websocket-configuration의 cryptoKeyReaderFactoryClassName 구성으로 암호화 리더가 구성된 경우에만.

메시지 게시 (Publish a message)

{
  "payload": "SGVsbG8gV29ybGQ=",
  "properties": {"key1": "value1", "key2": "value2"},
  "context": "1"
}
키 타입 필수? 설명
payload string 예 Base-64 인코딩된 페이로드
properties key-value pairs 아니요 애플리케이션 정의 속성
context string 아니요 애플리케이션 정의 요청 식별자
key string 아니요 파티셔닝된 토픽에서 어떤 파티션을 사용할지 결정
replicationClusters array 아니요 이름으로 지정된 이 클러스터 목록으로 복제 제한

성공 응답 예시 (Example success response)

{
   "result": "ok",
   "messageId": "CAAQAw==",
   "context": "1"
}

실패 응답 예시 (Example failure response)

{
   "result": "send-error:3",
   "errorMsg": "Failed to de-serialize from JSON",
   "context": "1"
}
키 타입 필수? 설명
result string 예 성공하면 ok, 실패하면 오류 메시지
messageId string 예 게시된 메시지에 할당된 메시지 ID
context string 아니요 애플리케이션 정의 요청 식별자

컨슈머 엔드포인트 (Consumer endpoint)

컨슈머 엔드포인트는 URL에 테넌트, 네임스페이스, 토픽뿐 아니라 서브스크립션도 지정해야 해요.

ws://broker-service-url:8080/ws/v2/consumer/persistent/:tenant/:namespace/:topic/:subscription

쿼리 파라미터 (Query param)

키 타입 필수? 설명
ackTimeoutMillis long 아니요 unacked 메시지의 타임아웃 설정 (기본: 0)
subscriptionType string 아니요 서브스크립션 타입: Exclusive, Failover, Shared, Key_Shared
receiverQueueSize int 아니요 컨슈머 수신 큐 크기 (기본: 1000)
consumerName string 아니요 컨슈머 이름
priorityLevel int 아니요 컨슈머의 우선순위 정의
maxRedeliverCount int 아니요 컨슈머의 maxRedeliverCount 정의 (기본: 0). Dead Letter Topic 기능 활성화.
deadLetterTopic string 아니요 컨슈머의 deadLetterTopic 정의 (기본: {topic}-{subscription}-DLQ). Dead Letter Topic 기능 활성화.
pullMode boolean 아니요 pull 모드 활성화 (기본: false). 아래 "Flow Control" 참고.
negativeAckRedeliveryDelay int 아니요 메시지가 부정 ack되었을 때 재전달 전 지연 시간(밀리초). 기본값 60000.
token string 아니요 인증 토큰. 브라우저 javascript 클라이언트에 사용돼요.

이 파라미터들(pullMode 제외)은 WebSocket 서비스의 내부 컨슈머에 적용돼요. 그래서 클라이언트가 WebSocket에서 소비하지 않더라도 메시지는 수신 큐에 들어오는 즉시 재전달 설정의 적용을 받아요.

메시지 수신 (Receive messages)

서버는 WebSocket 세션에 메시지를 푸시해요.

{
  "messageId": "CAMQADAA",
  "payload": "hvXcJvHW7kOSrUn17P2q71RA5SdiXwZBqw==",
  "properties": {},
  "publishTime": "2021-10-29T16:01:38.967-07:00",
  "redeliveryCount": 0,
  "encryptionContext": {
    "keys": {
      "client-rsa.pem": {
        "keyValue": "jEuwS+PeUzmCo7IfLNxqoj4h7txbLjCQjkwpaw5AWJfZ2xoIdMkOuWDkOsqgFmWwxiecakS6GOZHs94x3sxzKHQx9Oe1jpwBg2e7L4fd26pp+WmAiLm/ArZJo6JotTeFSvKO3u/yQtGTZojDDQxiqFOQ1ZbMdtMZA8DpSMuq+Zx7PqLo43UdW1+krjQfE5WD+y+qE3LJQfwyVDnXxoRtqWLpVsAROlN2LxaMbaftv5HckoejJoB4xpf/dPOUqhnRstwQHf6klKT5iNhjsY4usACt78uILT0pEPd14h8wEBidBz/vAlC/zVMEqiDVzgNS7dqEYS4iHbf7cnWVCn3Hxw==",
        "metadata": {}
      }
    },
    "param": "Tfu1PxVm6S9D3+Hk",
    "compressionType": "NONE",
    "uncompressedMessageSize": 0,
    "batchSize": {
      "empty": false,
      "present": true
    }
  }
}

WebSocket 컨슈머 응답의 파라미터는 다음과 같아요.

  • 일반 파라미터
키 타입 필수? 설명
messageId string 예 메시지 ID
payload string 예 Base-64 인코딩된 페이로드
publishTime string 예 게시 타임스탬프
redeliveryCount number 예 이 메시지가 이미 전달된 횟수
properties key-value pairs 아니요 애플리케이션 정의 속성
key string 아니요 프로듀서가 설정한 원래 라우팅 키
encryptionContext EncryptionContext 아니요 컨슈머가 수신 메시지를 복호화하는 데 사용할 수 있는 암호화 컨텍스트
param string 아니요 암호용 초기화 벡터 (Base64 인코딩)
batchSize string 아니요 메시지의 항목 수 (배치 메시지인 경우)
uncompressedMessageSize string 아니요 압축 전 메시지 크기
compressionType string 아니요 메시지 페이로드를 압축하는 데 사용된 알고리즘
  • encryptionContext 관련 파라미터
키 타입 필수? 설명
keys key-EncryptionKey pairs 예 key-EncryptionKey 쌍의 키는 암호화 키 이름이고, 값은 암호화 키 객체예요.
  • encryptionKey 관련 파라미터
키 타입 필수? 설명
keyValue string 예 암호화 키 (Base64 인코딩)
metadata key-value pairs 아니요 애플리케이션 정의 메타데이터

메시지 ack (Acknowledge the message)

컨슈머는 메시지의 성공적 처리를 ack해야 Pulsar 브로커가 그것을 삭제해요.

{
  "messageId": "CAAQAw=="
}
키 타입 필수? 설명
messageId string 예 처리된 메시지의 메시지 ID

메시지 부정 ack (Negatively acknowledge messages)

{
  "type": "negativeAcknowledge",
  "messageId": "CAAQAw=="
}
키 타입 필수? 설명
messageId string 예 처리된 메시지의 메시지 ID

흐름 제어 (Flow control)

푸시 모드 (Push Mode)

기본적으로(pullMode=false) 컨슈머 엔드포인트는 receiverQueueSize 파라미터를 사용해 내부 수신 큐를 크기 조정하고, WebSocket 클라이언트에 전달되는 unacknowledged 메시지 수를 제한해요. 이 모드에서 ack를 보내지 않으면, Pulsar WebSocket 서비스는 WebSocket 클라이언트에 보낸 receiverQueueSize 개의 unacked 메시지에 도달한 후 메시지 전송을 멈춰요.

풀 모드 (Pull Mode)

pullMode를 true로 설정하면 WebSocket 클라이언트는 Pulsar WebSocket 서비스가 더 많은 메시지를 보내도록 허용(permit) 명령을 보내야 해요.

{
  "type": "permit",
  "permitMessages": 100
}
키 타입 필수? 설명
type string 예 명령 타입. permit이어야 해요.
permitMessages int 예 허용할 메시지 수

이 모드에서는 다른 연결에서 메시지를 ack할 수 있어요.

토픽 끝 도달 확인 (Check if reach the end of topic)

컨슈머는 isEndOfTopic 요청을 보내 토픽 끝에 도달했는지 확인할 수 있어요.

요청:

{
  "type": "isEndOfTopic"
}
키 타입 필수? 설명
type string 예 명령 타입. isEndOfTopic이어야 해요.

응답:

{
   "endOfTopic": "true/false"
}

리더 엔드포인트 (Reader endpoint)

리더 엔드포인트는 URL에 테넌트, 네임스페이스, 토픽을 지정해야 해요.

ws://broker-service-url:8080/ws/v2/reader/persistent/:tenant/:namespace/:topic

쿼리 파라미터 (Query param)

키 타입 필수? 설명
readerName string 아니요 리더 이름
receiverQueueSize int 아니요 컨슈머 수신 큐 크기 (기본: 1000)
messageId int or enum 아니요 시작할 메시지 ID, earliest 또는 latest (기본: latest)
token string 아니요 인증 토큰. 브라우저 javascript 클라이언트에 사용돼요.

메시지 수신 (Receiving messages)

서버는 WebSocket 세션에 메시지를 푸시해요.

{
  "messageId": "CAAQAw==",
  "payload": "SGVsbG8gV29ybGQ=",
  "properties": {"key1": "value1", "key2": "value2"},
  "publishTime": "2016-08-30 16:45:57.785",
  "redeliveryCount": 4
}
키 타입 필수? 설명
messageId string 예 메시지 ID
payload string 예 Base-64 인코딩된 페이로드
publishTime string 예 게시 타임스탬프
redeliveryCount number 예 이 메시지가 이미 전달된 횟수
properties key-value pairs 아니요 애플리케이션 정의 속성
key string 아니요 프로듀서가 설정한 원래 라우팅 키

메시지 ack (Acknowledging the message)

WebSocket에서 리더는 메시지의 성공적 처리를 ack해야 Pulsar WebSocket 서비스가 보류 중인 메시지 수를 업데이트해요. ack를 보내지 않으면 Pulsar WebSocket 서비스는 pendingMessages 한도에 도달한 후 메시지 전송을 멈춰요.

{
  "messageId": "CAAQAw=="
}
키 타입 필수? 설명
messageId string 예 처리된 메시지의 메시지 ID

토픽 끝 도달 확인 (Check if reach the end of topic)

컨슈머는 isEndOfTopic 요청을 보내 토픽 끝에 도달했는지 확인할 수 있어요.

요청:

{
  "type": "isEndOfTopic"
}
키 타입 필수? 설명
type string 예 명령 타입. isEndOfTopic이어야 해요.

응답:

{
   "endOfTopic": "true/false"
}

오류 코드 (Error codes)

오류가 발생하면 서버는 다음 오류 코드로 WebSocket 세션을 닫아요.

오류 코드 오류 메시지
1 Failed to create producer
2 Failed to subscribe
3 Failed to deserialize from JSON
4 Failed to serialize to JSON
5 Failed to authenticate client
6 Client is not authorized
7 Invalid payload encoding
8 Unknown error

애플리케이션은 백오프 기간 후에 새 WebSocket 세션을 다시 수립할 책임이 있어요.

클라이언트 예시 (Client examples)

아래에서 Pulsar WebSocket API의 Python 및 Node.js 코드 예시를 찾을 수 있어요.

Python

이 예시는 websocket-client 패키지를 사용해요. pip으로 설치할 수 있어요.

pip install websocket-client

PyPI에서도 다운로드할 수 있어요.

Python 프로듀서 (Python producer)

다음은 Pulsar 토픽에 간단한 메시지를 보내는 Python 프로듀서 예시예요.

import websocket, base64, json

# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
enable_TLS = False
scheme = 'ws'
if enable_TLS:
    scheme = 'wss'

TOPIC = scheme + '://localhost:8080/ws/v2/producer/persistent/public/default/my-topic'
ws = websocket.create_connection(TOPIC)

# encode message
s = "Hello World"
firstEncoded = s.encode("UTF-8")
binaryEncoded = base64.b64encode(firstEncoded)
payloadString = binaryEncoded.decode('UTF-8')

# Send one message as JSON
ws.send(json.dumps({
    'payload' : payloadString,
    'properties': {
        'key1' : 'value1',
        'key2' : 'value2'
    },
    'context' : 5
}))

response =  json.loads(ws.recv())

if response['result'] == 'ok':
    print( 'Message published successfully')
else:
    print('Failed to publish message:', response)

ws.close()

Python 컨슈머 (Python consumer)

다음은 Pulsar 토픽을 듣고 메시지가 도착할 때마다 메시지 ID를 출력하는 Python 컨슈머 예시예요.

import websocket, base64, json

# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
enable_TLS = False
scheme = 'ws'
if enable_TLS:
    scheme = 'wss'

TOPIC = scheme + '://localhost:8080/ws/v2/consumer/persistent/public/default/my-topic/my-sub'
ws = websocket.create_connection(TOPIC)

while True:
    msg = json.loads(ws.recv())
    if not msg:
        break
    print( "Received: {} - payload: {}".format(msg, base64.b64decode(msg['payload'])))
    # Acknowledge successful processing
    ws.send(json.dumps({'messageId' : msg['messageId']}))

ws.close()

Python 리더 (Python reader)

다음은 Pulsar 토픽을 듣고 메시지가 도착할 때마다 메시지 ID를 출력하는 Python 리더 예시예요.

import websocket, base64, json

# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
enable_TLS = False
scheme = 'ws'
if enable_TLS:
    scheme = 'wss'

TOPIC = scheme + '://localhost:8080/ws/v2/reader/persistent/public/default/my-topic'
ws = websocket.create_connection(TOPIC)

while True:
    msg = json.loads(ws.recv())
    if not msg:
        break
    print ( "Received: {} - payload: {}".format(msg, base64.b64decode(msg['payload'])))
    # Acknowledge successful processing
    ws.send(json.dumps({'messageId' : msg['messageId']}))

ws.close()

Node.js

이 예시는 ws 패키지를 사용해요. npm으로 설치할 수 있어요.

npm install ws

Node.js 프로듀서 (Node.js producer)

다음은 Pulsar 토픽에 간단한 메시지를 보내는 Node.js 프로듀서 예시예요.

const WebSocket = require('ws');

// If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
const enableTLS = false;
const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/producer/persistent/public/default/my-topic`;
const ws = new WebSocket(topic);

var message = {
  "payload" : new Buffer("Hello World").toString('base64'),
  "properties": {
    "key1" : "value1",
    "key2" : "value2"
  },
  "context" : "1"
};

ws.on('open', function() {
  // Send one message
  ws.send(JSON.stringify(message));
});

ws.on('message', function(message) {
  console.log('received ack: %s', message);
});

Node.js 컨슈머 (Node.js consumer)

다음은 위 프로듀서가 사용한 것과 같은 토픽을 듣는 Node.js 컨슈머 예시예요.

const WebSocket = require('ws');

// If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
const enableTLS = false;
const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/consumer/persistent/public/default/my-topic/my-sub`;
const ws = new WebSocket(topic);

ws.on('message', function(message) {
    var receiveMsg = JSON.parse(message);
    console.log('Received: %s - payload: %s', message, new Buffer(receiveMsg.payload, 'base64').toString());
    var ackMsg = {"messageId" : receiveMsg.messageId};
    ws.send(JSON.stringify(ackMsg));
});

Node.js 리더 (NodeJS reader)

const WebSocket = require('ws');

// If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
const enableTLS = false;
const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/reader/persistent/public/default/my-topic`;
const ws = new WebSocket(topic);

ws.on('message', function(message) {
    var receiveMsg = JSON.parse(message);
    console.log('Received: %s - payload: %s', message, new Buffer(receiveMsg.payload, 'base64').toString());
    var ackMsg = {"messageId" : receiveMsg.messageId};
    ws.send(JSON.stringify(ackMsg));
});

더 알아보기 (Learn more)