sparrow/pkg/queue/queue_callback.go

package queue

import (
	"sparrow/pkg/protocol"
	"sync/atomic"
)

type Callback interface {
	OnSuccess()
	OnFailure(err error)
}

type MsgCallbackWrapper struct {
	msgCallback protocol.IMessageCallBack
}

func NewMsgCallbackWrapper(cb protocol.IMessageCallBack) *MsgCallbackWrapper {
	return &MsgCallbackWrapper{msgCallback: cb}
}

func (m *MsgCallbackWrapper) OnSuccess() {
	m.OnSuccess()
}

func (m *MsgCallbackWrapper) OnFailure(err error) {
	m.msgCallback.OnFailure(err)
}

type MultipleMsgCallbackWrapper struct {
	callbackCount int32
	msgCallBack   protocol.IMessageCallBack
}

func NewMultipleMsgCallbackWrapper(count int32, callback protocol.IMessageCallBack) *MultipleMsgCallbackWrapper {
	return &MultipleMsgCallbackWrapper{
		callbackCount: count,
		msgCallBack:   callback,
	}
}

func (m *MultipleMsgCallbackWrapper) OnSuccess() {
	v := atomic.AddInt32(&m.callbackCount, -1)
	if v <= 0 {
		m.msgCallBack.OnSuccess()
	}
}

func (m *MultipleMsgCallbackWrapper) OnFailure(err error) {
	m.msgCallBack.OnFailure(err)
}