mirror of
https://github.com/duanhf2012/origin.git
synced 2026-02-03 22:45:13 +08:00
92 lines
2.5 KiB
Go
92 lines
2.5 KiB
Go
package messagequeueservice
|
||
|
||
import (
|
||
"github.com/duanhf2012/origin/v2/log"
|
||
"github.com/duanhf2012/origin/v2/rpc"
|
||
|
||
"sync"
|
||
)
|
||
|
||
// 订阅器
|
||
type Subscriber struct {
|
||
customerLocker sync.RWMutex
|
||
mapCustomer map[string]*CustomerSubscriber
|
||
queue MemoryQueue
|
||
dataPersist QueueDataPersist //对列数据处理器
|
||
queueWait *sync.WaitGroup
|
||
}
|
||
|
||
func (ss *Subscriber) Init(memoryQueueCap int32) {
|
||
ss.queue.Init(memoryQueueCap)
|
||
ss.mapCustomer = make(map[string]*CustomerSubscriber, 5)
|
||
}
|
||
|
||
func (ss *Subscriber) PushTopicDataToQueue(topic string, topics []TopicData) {
|
||
for i := 0; i < len(topics); i++ {
|
||
ss.queue.Push(&topics[i])
|
||
}
|
||
}
|
||
|
||
func (ss *Subscriber) PersistTopicData(topic string, topics []TopicData, retryCount int) ([]TopicData, []TopicData, bool) {
|
||
return ss.dataPersist.PersistTopicData(topic, topics, retryCount)
|
||
}
|
||
|
||
func (ss *Subscriber) TopicSubscribe(rpcHandler rpc.IRpcHandler, subScribeType rpc.SubscribeType, subscribeMethod SubscribeMethod, fromNodeId string, callBackRpcMethod string, topic string, customerId string, StartIndex uint64, oneBatchQuantity int32) error {
|
||
//取消订阅时
|
||
if subScribeType == rpc.SubscribeType_Unsubscribe {
|
||
ss.UnSubscribe(customerId)
|
||
return nil
|
||
} else {
|
||
ss.customerLocker.Lock()
|
||
customerSubscriber, ok := ss.mapCustomer[customerId]
|
||
if ok == true {
|
||
//已经订阅过，则取消订阅
|
||
customerSubscriber.UnSubscribe()
|
||
delete(ss.mapCustomer, customerId)
|
||
}
|
||
|
||
//不存在，则订阅
|
||
customerSubscriber = &CustomerSubscriber{}
|
||
ss.mapCustomer[customerId] = customerSubscriber
|
||
ss.customerLocker.Unlock()
|
||
|
||
err := customerSubscriber.Subscribe(rpcHandler, ss, topic, subscribeMethod, customerId, fromNodeId, callBackRpcMethod, StartIndex, oneBatchQuantity)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
if ok == true {
|
||
log.Info("repeat subscription for customer ", customerId)
|
||
} else {
|
||
log.Info("subscription for customer ", customerId)
|
||
}
|
||
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (ss *Subscriber) UnSubscribe(customerId string) {
|
||
ss.customerLocker.RLocker()
|
||
defer ss.customerLocker.RUnlock()
|
||
|
||
customerSubscriber, ok := ss.mapCustomer[customerId]
|
||
if ok == false {
|
||
log.SWarning("failed to unsubscribe customer " + customerId)
|
||
return
|
||
}
|
||
|
||
customerSubscriber.UnSubscribe()
|
||
}
|
||
|
||
func (ss *Subscriber) removeCustomer(customerId string, cs *CustomerSubscriber) {
|
||
|
||
ss.customerLocker.Lock()
|
||
//确保删掉是当前的关系。有可能在替换订阅时，将该customer替换的情况
|
||
customer, _ := ss.mapCustomer[customerId]
|
||
if customer == cs {
|
||
delete(ss.mapCustomer, customerId)
|
||
}
|
||
ss.customerLocker.Unlock()
|
||
}
|