Go 채널 분배기

Link

• https://github.com/p9595jh/chdist
• https://pkg.go.dev/github.com/p9595jh/chdist

Channel Distributor

Go언어 채널에 여러 구독자를 붙일 수 있는 패키지.
제네릭을 사용하기 때문에 1.18 이상의 버전이 필요하다.

설명

Go언어 채널은 한 번에 하나의 수신만이 가능하다. chdist는 pub-sub 모델을 구현함으로써 일종의 메시지큐 시스템과 같은 이용이 가능하다. 동기식과 비동기식이 모두 제공되며 일반적인 채널을 다루듯이 사용할 수 있다.

Distributor

Distributor는 생산자를 담당한다. Distributor의 생성자는 제네릭 타입과 그 타입에 해당하는 채널을 파라미터로 받아서 새로운 객체를 생성하며, 동기 혹은 비동기 방식으로 구독이 가능하다.

channel := make(chan int)
distributor := chdist.NewDistributor[int](channel)

소재가 되는 채널을 통해서뿐만 아니라 직접 데이터 전송도 가능하다.

channel <- 5
// or
distributor.In() <- 5

다중 구독도 물론 가능하다.

// sync
distributor.Subscribe(func(value int) {
    fmt.Println(value)
})

// async
for value := range distributor.AsyncSubscribe.Out() {
    fmt.Println(value)
}

Close() 메소드를 이용해서 종료할 수 있다.

distributor.Close()

Subscription

distributor가 구독될 때 Subscription 객체를 리턴한다. 이 객체를 통해 구독을 끊거나, 구독이 살아있는지 확인하는 등의 행동을 할 수 있다.

subscription := distributor.Subscribe(func(value int) { /* logics */ })
if subscription.IsAlive() {
    subscription.Close()
}

AsyncSubscription은 일반적인 채널을 이용하듯이 쓸 수 있다.

asyncSubscription := distributor.AsyncSubscribe()
fmt.Println(<-asyncSubscription.Out())

JsonDistributor

JsonDistributor는 JSON 데이터를 다룰 수 있다. 직렬화된 JSON이 입력으로 들어가며 언마샬된 타입을 제네릭으로 받는다.
문자열과 byte 슬라이스([]byte)로 된 JSON 데이터를 지원한다(각 NewJsonStringDistributor와 NewJsonBytesDistributor로 생성 가능하다).
encoding/json 패키지를 이용하기 때문에 언마샬 도중 출력된 에러를 같이 리턴할 수 있도록 Item이라는 구조체를 이용한다.

type Item[T any] struct {
	Value T
	Error error
}

JsonStringDistributor의 생성자는 문자열 타입 Distributor를, JsonBytesDistributor는 byte 슬라이스([]byte) 타입의 Distributor를 파라미터로 받는다.

type Sample struct {
    Num int `json:"num"`
}

channel := make(chan string)

jsonStringDistributor := chdist.NewJsonStringDistributor[Sample](
    chdist.NewDistributor[string](channel),
)

go func() {
    for {
        channel <- `{"num":123}`
        time.Sleep(time.Second * 5)
    }
}()

jsonStringDistributor.Subscribe(func(item chdist.Item[Sample]) {
    fmt.Println(item.Value) // type of `Sample`
    fmt.Println(item.Error) // if error occured while unmarshaling, it goes to
})

사용 방법

go get github.com/p9595jh/chdist

import "github.com/p9595jh/chdist"