[nats] Request-Reply

Request-Reply 패턴은 기존의 메모리 큐 시스템과 차별화되는 nats의 pub/sub 확장기능 중 하나다.

이를 이용하면 웹서버처럼 요청을 보내고 응답을 받아오는 식의 처리가 가능하다.
동기적으로 사용할 수도, 비동기적으로 사용할 수도 있다.

먼저 CLI를 사용해서 응답 형태를 만들어보겠다.
아래는 help.please라는 요청이 날라올 경우 'OK, I CAN HELP!!!'를 응답으로 반환해주는 응답이다.

nats reply --server nats://localhost:4222 help.please 'OK, I CAN HELP!'

이러면 일단 대기중이 될 것이다.

그리고 다음과 같이 요청을 날린다.

nats request --server nats://localhost:4222 help.please 'I need help!'

그러면 reply에서 보내준 메세지가 request 측에서 보일 것이고

request에서 보내준 메세지도 reply 측에서 보일 것이다.

그리고 이건 pub/sub과 다르게, reply가 여러개라도 하나만 선정돼서 처리된다.
아래는 한 주제에 대해 2가지 "reply"를 정의하고 request를 날렸을 때 발생하는 동작이다.

그냥 랜덤으로 하나씩 호출해서 받아온다.

이걸 이용하면 HTTP 서버와 비스무리한 기능들을 구현할 수도 있다.

아래는 덧셈을 수행해주는 "add"라는 주제를 만들고 사용하는 간단한 예제다.
언어는 Go다. 특별한 이유는 없다.

reply 프로세스는 요청이 왔을 경우의 응답을 처리한다.
메세지에서 인자 목록을 쉼표 구분으로 받고, 더해서 결과를 알려준다.

package main

import (
	"fmt"
	"strconv"
	"strings"

	"github.com/nats-io/nats.go"
)

func main() {
	client, _ := nats.Connect(nats.DefaultURL)

	// Reply
	client.Subscribe("add", func(m *nats.Msg) {
		input := string(m.Data)
		operands := strings.Split(input, ",")
		lhs, lhsErr := strconv.ParseInt(operands[0], 10, 64)
		rhs, rhsErr := strconv.ParseInt(operands[1], 10, 64)

		if lhsErr != nil || rhsErr != nil {
			client.Publish(m.Reply, []byte("I can't help!"))
			return
		}

		sum := lhs + rhs
		pubMessage := fmt.Sprintf("Sum is %d", sum)
		pubError := client.Publish(m.Reply, []byte(pubMessage))

		if pubError != nil {
			fmt.Println(pubError)
		}
	})

	fmt.Println("replier is running")

	for {
	}
}

request 프로세스는 이에 비해 간단하다.
주제명과 메세지, 타임아웃을 전달한다.
타임아웃 시간동안은 reply가 구성되어있지 않거나 답이 오지 않더라도 기다린다.

package main

import (
	"fmt"
	"time"

	"github.com/nats-io/nats.go"
)

func main() {
	client, _ := nats.Connect(nats.DefaultURL)

	message, err := client.Request("add", []byte("10,20"), 100*time.Millisecond)

	fmt.Println("replier received a reply")
	fmt.Println(string(message.Data))

	if err != nil {
		fmt.Println(err)
		panic(err)
	}
}

reply를 실행해두고

메세지가 이렇게 날라오면 잘 된 것이다.

참조
요청-회신 - NATS 문서