mirror of
https://gitee.com/willfree/mlsr.git
synced 2026-10-04 19:42:02 +08:00
101 lines
3.1 KiB
Go
101 lines
3.1 KiB
Go
// Package communication
|
|
// @Author: zhengqi Wu
|
|
// @Description:
|
|
// @Version: 0.1.0
|
|
// @Date: 2022/8/10 16:07
|
|
// @Copyright: MIN-Group；国家重大科技基础设施——未来网络北大实验室；深圳市信息论与未来网络重点实验室
|
|
//
|
|
|
|
package communication
|
|
|
|
import (
|
|
"MINSync"
|
|
"errors"
|
|
"minlib/common"
|
|
"minlib/component"
|
|
"minlib/logicface"
|
|
"time"
|
|
)
|
|
|
|
// SyncUpdateCallback 当同步协议获取到其它节点的更新数据时
|
|
// 所调用的回调函数
|
|
//
|
|
type SyncUpdateCallback = func(updateName *component.Identifier, seqNo uint64, incomingFaceId uint64)
|
|
|
|
// SyncProtocolAdapter 同步协议的简单封装
|
|
//
|
|
// 当前默认使用MINSync同步协议，故将Producer设置为MINSync的FullProducer
|
|
//
|
|
type SyncProtocolAdapter struct {
|
|
/////////// 私有属性 //////////
|
|
// syncUpdateCallback 当同步协议获取到其它节点的更新数据时所调用的自定义函数
|
|
syncUpdateCallback SyncUpdateCallback
|
|
// minSyncLogic MINSync全同步逻辑实例
|
|
minSyncLogic *MINSync.FullProducer
|
|
}
|
|
|
|
// AddUserNode 为MINSync 全同步生产者添加数据前缀
|
|
//
|
|
// 参数说明：
|
|
//
|
|
// userPrefix: 待添加的数据前缀
|
|
//
|
|
func (s *SyncProtocolAdapter) AddUserNode(userPrefix *component.Identifier) {
|
|
s.minSyncLogic.AddUserNode(userPrefix)
|
|
}
|
|
|
|
// PublishUpdate 向其它路由节点发布更新
|
|
//
|
|
// MLSR 有着自己一套的序列号管理规则，因此每次发布更新都将强制更新生产者的序列号
|
|
//
|
|
func (s *SyncProtocolAdapter) PublishUpdate(userPrefix *component.Identifier, seq uint64) {
|
|
s.minSyncLogic.PublishName(userPrefix, int64(seq))
|
|
}
|
|
|
|
// OnMINSyncUpdate 每当MINSync获取到新同步数据时便会调用的回调函数
|
|
//
|
|
// 该函数将新同步数据封装成独立的更新并向上交付给自定义的更新处理函数syncUpdateCallback
|
|
//
|
|
// 参数说明：
|
|
//
|
|
// updates: MissingDataInfo的列表
|
|
//
|
|
func (s *SyncProtocolAdapter) OnMINSyncUpdate(updates []*MINSync.MissingDataInfo) {
|
|
common.LogInfo("Received PSync update event")
|
|
|
|
for _, update := range updates {
|
|
s.syncUpdateCallback(update.Prefix, update.HighSeq, update.IncomingFace)
|
|
}
|
|
}
|
|
|
|
// NewSyncProtocolAdapter SyncProtocolAdapter的工厂方法
|
|
//
|
|
func NewSyncProtocolAdapter(face *logicface.DummyClientLogicFace,
|
|
syncPrefix, userPrefix *component.Identifier,
|
|
syncInterestLifetime time.Duration,
|
|
syncUpdateCallback SyncUpdateCallback) (*SyncProtocolAdapter, error) {
|
|
|
|
if face == nil || syncPrefix == nil || userPrefix == nil || syncUpdateCallback == nil {
|
|
return nil, errors.New("无效的输入参数")
|
|
}
|
|
common.LogInfo("Using MINSync")
|
|
|
|
syncProtocolAdapter := new(SyncProtocolAdapter)
|
|
syncProtocolAdapter.syncUpdateCallback = syncUpdateCallback
|
|
// 根据PSync源码来说，默认将Sync Data的生命周期设为SYNC_REPLY_FRESHNESS
|
|
// 魔数 80 意味着两个路由节点的数据前缀差集最大在80条左右
|
|
fullProducer, err := MINSync.NewFullProducer(
|
|
80,
|
|
face,
|
|
syncPrefix, userPrefix,
|
|
syncProtocolAdapter.OnMINSyncUpdate,
|
|
syncInterestLifetime, MINSync.SYNC_REPLY_FRESHNESS)
|
|
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
syncProtocolAdapter.minSyncLogic = fullProducer
|
|
return syncProtocolAdapter, nil
|
|
}
|