重写Hello协议,增加处理来自其它路由器的兴趣包的代码(解析兴趣包,然后给一个data包做应答)

This commit is contained in:
free will committed 2022-07-14 23:27:57 +08:00
1 parent 9514e2e3c4
commit c2c9b7cf2c
6 files changed
+655 -2

No files matched your search

+2 -2
View File
@@ -355,8 +355,8 @@ func (mcp *MlsrConfigParameters) GetInterestRetryNumber() {
// todo
}
func (mcp *MlsrConfigParameters) GetInterestResendTime() {
// todo
func (mcp *MlsrConfigParameters) GetInterestResendTime() uint32 {
return mcp.mlsrconf.HelloTimeout
}
func (mcp *MlsrConfigParameters) GetInfoInterestInterval() {
+36
View File
@@ -0,0 +1,36 @@
// Package hello
// @Author: Wang Feng
// @Description: Hello协议的辅助函数
// @Version: 0.1.0
// @Date: 2022/7/14 22:20
// @Copyright: MIN-Group;国家重大科技基础设施——未来网络北大实验室;深圳市信息论与未来网络重点实验室
//
package hello
import (
"minlib/component"
"mlsr/utils"
)
//
// GetPrefixForHelloProtocol
// @Description: 将一个多层级prefix(即一个可能以多个/分割的字符串),转变为一个单层级prefix的内容(即其形态中没有/)
// @param identifier
//
func GetSinglePrefixForHelloProtocol(identifier *component.Identifier) string {
return utils.Base62StdEncoding.EncodeToString([]byte(identifier.ToUri()))
}
//
// GetRawPrefixForHelloProtocol
// @Description: 将单层级prefix的内容(即去掉首个/的前缀),转为其原始形态
// @return string
//
func GetRawPrefixForHelloProtocol(prefixContent string) (string, error) {
decodeRes, err := utils.Base62StdEncoding.DecodeString(prefixContent)
if err != nil {
return "", err
}
return string(decodeRes), nil
}
+104
View File
@@ -0,0 +1,104 @@
// Package hello
// @Author: Pei Xinyuan
// @Description:
// @Version: 0.1.0
// @Date: 2022/7/14 14:49
// @Copyright: MIN-Group;国家重大科技基础设施——未来网络北大实验室;深圳市信息论与未来网络重点实验室
//
package hello
import (
common2 "minlib/common"
"minlib/component"
"minlib/logicface"
"minlib/packet"
"minlib/security"
"mlsr/common"
"mlsr/lsa"
"mlsr/lsdb"
"mlsr/route"
)
// 用于hello兴趣包的名称的组件
const INFO_COMPONENT = "INFO"
const MLSR_COMPONENT = "mlsr"
// 正常收到数据包的触发函数
type DataCallback = func(interest *packet.Interest, data *packet.Data)
// 收到NACK的触发函数
type NackCallback = func(interest *packet.Interest, nack *packet.Nack)
// 超时未响应的触发函数
type TimeoutCallback = func(interest *packet.Interest)
//type OptionalNack = *packet.Nack
type HelloProtocol struct {
m_face *logicface.LogicFace
m_scheduler *lsdb.MlsrScheduler
m_keychain *security.KeyChain
mlsrConfig *common.MlsrConfig
m_routingTable *route.RoutingTable
m_lsdb *lsdb.Lsdb
m_adjacencyList *lsa.AdjLsaAdjacenctList
// 收包池
receivePacketQue chan *packet.Interest // 从其它路由器收包后,放入该队列
state bool // 程序生成状态, true 为正常运行, false 为关闭状态
}
//
// Init
// @Description: Hello协议-初始化
// @receiver hp
// @param face
// @param scheduler
// @param keychain
// @param mlsrConfig
// @param routingTable
// @param mlsdb
//
func (hp *HelloProtocol) Init(face *logicface.LogicFace,
scheduler *lsdb.MlsrScheduler,
keychain *security.KeyChain, mlsrConfig *common.MlsrConfig,
routingTable *route.RoutingTable, mlsdb *lsdb.Lsdb) {
hp.m_face = face
hp.m_scheduler = scheduler
hp.m_keychain = keychain
hp.mlsrConfig = mlsrConfig
hp.m_routingTable = routingTable
hp.m_lsdb = mlsdb
hp.m_adjacencyList = mlsrConfig.GetAdjacencys()
// todo mlsr大联调时应取消以下代码的注释
// 构造接收hello兴趣包的前缀
name, err := component.CreateIdentifierByString(
hp.mlsrConfig.GetRouterPrefix().ToUri() + "/" + MLSR_COMPONENT + "/" + INFO_COMPONENT)
if err != nil {
common2.LogFatal("HelloProtocol Setting interest filter for Hello interest error, interest generate wrong.")
return
}
// 设置监听前缀
err = hp.m_face.RegisterIdentifier(name, 3000)
if err != nil {
common2.LogFatal("HelloProtocol Setting interest filter for Hello interest error, identifier register wrong.")
return
}
// 开一个线程去接收前缀为hello协议所监听前缀的兴趣包,并进行处理
hp.receivePacketQue = make(chan *packet.Interest, 1000)
hp.state = true
// 启动处理接收到的包的协程
go hp.dealWithRecvPktRoutine()
// 启动从客户端接收包的协程
go hp.recvPacketFromClientRoutine()
}
func (hp *HelloProtocol) expressInterest(interestName *component.Identifier, seconds uint32) {
common2.LogDebug("HelloProtocol Expressing Interest: " + interestName.ToUri())
interest := packet.NewInterest()
interest.SetName(interestName)
interest.SetInterestLifeTime(2000)
// todo 这里需要实现的是,将兴趣包通过face发出,并对超时、nack、正常应答三种情况进行处理
}
+141
View File
@@ -0,0 +1,141 @@
// Package hello
// @Author: Wang Feng
// @Description:
// @Version: 0.1.0
// @Date: 2022/7/14 22:01
// @Copyright: MIN-Group;国家重大科技基础设施——未来网络北大实验室;深圳市信息论与未来网络重点实验室
//
package hello
import (
"errors"
"log"
common2 "minlib/common"
"minlib/component"
"minlib/packet"
"mlsr/lsa"
"unsafe"
)
//
// @Description: 接收MIN包的协程
// @receiver v
//
func (v *HelloProtocol) recvPacketFromClientRoutine() {
for v.state {
err := v.recvPacketFromClient()
if err != nil {
log.Fatal(err)
}
}
}
//
// @Description: 等待接收一个MIN包,并放入管道
// @receiver v
// @return error
//
func (v *HelloProtocol) recvPacketFromClient() error {
interest, err := v.m_face.ReceiveInterest(-1)
common2.LogDebug("HelloProtocol recv interest packet, the name is: ", interest.GetName().ToUri())
if err != nil {
return err
}
v.receivePacketQue <- interest
return nil
}
//
// @Description: 处理收到的MIN包的协程
// @receiver v
//
func (v *HelloProtocol) dealWithRecvPktRoutine() {
for v.state {
err := v.dealWithRecvPkt()
if err != nil {
log.Println(err)
break
}
}
}
//
// @Description: 从管道读取并处理一个MIN包
// @receiver v
// @return error
//
func (v *HelloProtocol) dealWithRecvPkt() error {
interest, ok := <-v.receivePacketQue
if !ok {
common2.LogDebug("HelloProtocol get interest packet from receivePacketQue error")
return errors.New("HelloProtocol get interest packet from receivePacketQue error")
}
// 取出兴趣包倒数第二个组件,对其进行校验
infoComponent, err := interest.GetName().Get(-2)
if err != nil {
common2.LogDebug("HelloProtocol deal with interest error, get(-2) err ", err)
return err
}
if infoComponent.ToUri() != INFO_COMPONENT {
common2.LogDebug("HelloProtocol deal with interest error, INFO_COMPONENT not found or interestName ", err)
return err
}
// 取出兴趣包最后一个组件,其为该兴趣包所包含的RouterPrefix信息。
prefixComponent, err := interest.GetName().Get(-1)
if err != nil {
common2.LogDebug("HelloProtocol deal with interest error, get(-1) err", err)
return err
}
routerPrefix, err := GetRawPrefixForHelloProtocol(prefixComponent.ToUri())
if err != nil {
common2.LogDebug("HelloProtocol deal with interest error, GetRawPrefixForHelloProtocol err ", err)
return err
}
routerIdentifier, err := component.CreateIdentifierByString(routerPrefix)
if err != nil {
common2.LogDebug("HelloProtocol deal with interest error, create identifier from prefix err ", err)
return err
}
if v.m_adjacencyList.IsNeighbor(routerIdentifier) {
// 构造Data,以向兴趣包做应答
data := packet.NewData()
data.SetName(interest.GetName())
data.SetFreshnessPeriod(10 * 1000)
// todo 安全性:对该data包进行签名
// 使用face对兴趣包进行响应
err = v.m_face.SendData(data)
if err != nil {
common2.LogDebug("HelloProtocol deal with interest error, send data error ", err)
return err
}
// 找到该邻接路由器的状态信息
adjInfo := v.m_adjacencyList.FindAdjacencyByRouterIdentifier(routerIdentifier)
// 如果该路由器状态在本机这里被记为未激活,则向其发送hello包,来确认其已激活
if adjInfo.Status == lsa.STATUS_INACTIVE {
// interest name: /<neighbor>/NLSR/INFO/<router>
name, err := component.CreateIdentifierByString(routerPrefix +
"/" + MLSR_COMPONENT + "/" + INFO_COMPONENT +
"/" + GetSinglePrefixForHelloProtocol(v.mlsrConfig.GetRouterPrefix()))
if err != nil {
common2.LogDebug("HelloProtocol deal with interest error, hello name create wrong.")
return err
}
// 将data发出去
v.expressInterest(name, v.mlsrConfig.GetInterestResendTime())
}
}
return nil
}
func stringToBytes(str string) []byte {
return *(*[]byte)(unsafe.Pointer(&str))
}
func byteToString(bytes []byte) string {
return *(*string)(unsafe.Pointer(&bytes))
}
+337
View File
@@ -0,0 +1,337 @@
// Package hello
// @Author: Wang Feng
// @Description: base62编解码。源代码:https://github.com/lytics/base62
// @Version: 0.1.0
// @Date: 2022/7/14 22:20
// @Copyright: MIN-Group;国家重大科技基础设施——未来网络北大实验室;深圳市信息论与未来网络重点实验室
//
// heavily influcened by http://golang.org/src/pkg/encoding/base64/base64.go
// Package base62 implements base62 encoding
package utils
import (
"errors"
"fmt"
"io"
)
// Encodings
type Encoding struct {
encode string
decodeMap [256]byte
}
const encodeStd = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz-0123456789." //"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
// NewEncoding returns a new Encoding defined by the given alphabet,
// which must be a 62-byte string.
func NewEncoding(encoder string) *Encoding {
e := new(Encoding)
e.encode = encoder
for i := 0; i < len(e.decodeMap); i++ {
e.decodeMap[i] = 0xFF
}
for i := 0; i < len(encoder); i++ {
e.decodeMap[encoder[i]] = byte(i)
}
return e
}
var Base62StdEncoding = NewEncoding(encodeStd)
/*
* Encoder
*/
// Encode encodes src using the encoding enc, writing
// EncodedLen(len(src)) bytes to dst.
//
// The encoding pads the output to a multiple of 4 bytes,
// so Encode is not appropriate for use on individual blocks
// of a large data stream. Use NewEncoder() instead.
func (enc *Encoding) Encode(dst, src []byte) {
if len(src) == 0 {
return
}
for len(src) > 0 {
dst[0] = 0
dst[1] = 0
dst[2] = 0
dst[3] = 0
// Unpack 4x 6-bit source blocks into a 4 byte
// destination quantum
switch len(src) {
default:
dst[3] |= src[2] & 0x3F
dst[2] |= src[2] >> 6
fallthrough
case 2:
dst[2] |= (src[1] << 2) & 0x3F
dst[1] |= src[1] >> 4
fallthrough
case 1:
dst[1] |= (src[0] << 4) & 0x3F
dst[0] |= src[0] >> 2
}
// Encode 6-bit blocks using the base62 alphabet
for j := 0; j < 4; j++ {
dst[j] = enc.encode[dst[j]]
}
// Pad the final quantum
if len(src) < 3 {
dst[3] = '+'
if len(src) < 2 {
dst[2] = '+'
}
break
}
src = src[3:]
dst = dst[4:]
}
}
// EncodeToString returns the base62 encoding of src.
func (enc *Encoding) EncodeToString(src []byte) string {
buf := make([]byte, enc.EncodedLen(len(src)))
enc.Encode(buf, src)
return string(buf)
}
type encoder struct {
err error
enc *Encoding
w io.Writer
buf [3]byte // buffered data waiting to be encoded
nbuf int // number of bytes in buf
out [1024]byte // output buffer
}
func (e *encoder) Write(p []byte) (n int, err error) {
if e.err != nil {
return 0, e.err
}
// Leading fringe.
if e.nbuf > 0 {
var i int
for i = 0; i < len(p) && e.nbuf < 3; i++ {
e.buf[e.nbuf] = p[i]
e.nbuf++
}
n += i
p = p[i:]
if e.nbuf < 3 {
return
}
e.enc.Encode(e.out[0:], e.buf[0:])
if _, e.err = e.w.Write(e.out[0:4]); e.err != nil {
return n, e.err
}
e.nbuf = 0
}
// Large interior chunks.
for len(p) >= 3 {
nn := len(e.out) / 4 * 3
if nn > len(p) {
nn = len(p)
}
nn -= nn % 3
if nn > 0 {
e.enc.Encode(e.out[0:], p[0:nn])
if _, e.err = e.w.Write(e.out[0 : nn/3*4]); e.err != nil {
return n, e.err
}
}
n += nn
p = p[nn:]
}
// Trailing fringe.
for i := 0; i < len(p); i++ {
e.buf[i] = p[i]
}
e.nbuf = len(p)
n += len(p)
return
}
// Close flushes any pending output from the encoder.
// It is an error to call Write after calling Close.
func (e *encoder) Close() error {
// If there's anything left in the buffer, flush it out
if e.err == nil && e.nbuf > 0 {
e.enc.Encode(e.out[0:], e.buf[0:e.nbuf])
e.nbuf = 0
_, e.err = e.w.Write(e.out[0:4])
}
return e.err
}
// NewEncoder returns a new base62 stream encoder. Data written to
// the returned writer will be encoded using enc and then written to w.
// Base62 encodings operate in 4-byte blocks; when finished
// writing, the caller must Close the returned encoder to flush any
// partially written blocks.
func NewEncoder(enc *Encoding, w io.Writer) io.WriteCloser {
return &encoder{enc: enc, w: w}
}
// EncodedLen returns the length in bytes of the base62 encoding
// of an input buffer of length n.
func (enc *Encoding) EncodedLen(n int) int { return (n + 2) / 3 * 4 }
/*
* Decoder
*/
type CorruptInputError error
// func (e CorruptInputError) String() error {
// return "illegal base62 data at input byte " + strconv.FormatInt(int64(e), 10)
// }
// decode is like Decode but returns an additional 'end' value, which
// indicates if end-of-message padding was encountered and thus any
// additional data is an error. decode also assumes len(src)%4==0,
// since it is meant for internal use.
func (enc *Encoding) decode(dst, src []byte) (n int, end bool, err error) {
for i := 0; i < len(src)/4 && !end; i++ {
// Decode quantum using the base62 alphabet
var dbuf [4]byte
dlen := 4
dbufloop:
for j := 0; j < 4; j++ {
in := src[i*4+j]
if in == '+' && j >= 2 && i == len(src)/4-1 {
// We've reached the end and there's
// padding
if src[i*4+3] != '+' {
return n, false, errors.New(fmt.Sprintf("%d", i*4+2))
}
dlen = j
end = true
break dbufloop
}
dbuf[j] = enc.decodeMap[in]
if dbuf[j] == 0xFF {
return n, false, errors.New(fmt.Sprintf("%d", i*4+j))
}
}
// Pack 4x 6-bit source blocks into 3 byte destination
// quantum
switch dlen {
case 4:
dst[i*3+2] = dbuf[2]<<6 | dbuf[3]
fallthrough
case 3:
dst[i*3+1] = dbuf[1]<<4 | dbuf[2]>>2
fallthrough
case 2:
dst[i*3+0] = dbuf[0]<<2 | dbuf[1]>>4
}
n += dlen - 1
}
return n, end, nil
}
// Decode decodes src using the encoding enc. It writes at most
// DecodedLen(len(src)) bytes to dst and returns the number of bytes
// written. If src contains invalid base62 data, it will return the
// number of bytes successfully written and CorruptInputError.
func (enc *Encoding) Decode(dst, src []byte) (n int, err error) {
pad := len(src) % 4
if pad != 0 {
for i := 0; i < pad; i++ {
src = append(src, '+')
}
}
n, _, err = enc.decode(dst, src)
return
}
// DecodeString returns the bytes represented by the base62 string s.
func (enc *Encoding) DecodeString(s string) ([]byte, error) {
dbuf := make([]byte, enc.DecodedLen(len(s)))
n, err := enc.Decode(dbuf, []byte(s))
return dbuf[:n], err
}
type decoder struct {
err error
enc *Encoding
r io.Reader
end bool // saw end of message
buf [1024]byte // leftover input
nbuf int
out []byte // leftover decoded output
outbuf [1024 / 4 * 3]byte
}
func (d *decoder) Read(p []byte) (n int, err error) {
if d.err != nil {
return 0, d.err
}
// Use leftover decoded output from last read.
if len(d.out) > 0 {
n = copy(p, d.out)
d.out = d.out[n:]
return n, nil
}
// Read a chunk.
nn := len(p) / 3 * 4
if nn < 4 {
nn = 4
}
if nn > len(d.buf) {
nn = len(d.buf)
}
nn, d.err = io.ReadAtLeast(d.r, d.buf[d.nbuf:nn], 4-d.nbuf)
d.nbuf += nn
if d.nbuf < 4 {
return 0, d.err
}
// Decode chunk into p, or d.out and then p if p is too small.
nr := d.nbuf / 4 * 4
nw := d.nbuf / 4 * 3
if nw > len(p) {
nw, d.end, d.err = d.enc.decode(d.outbuf[0:], d.buf[0:nr])
d.out = d.outbuf[0:nw]
n = copy(p, d.out)
d.out = d.out[n:]
} else {
n, d.end, d.err = d.enc.decode(p, d.buf[0:nr])
}
d.nbuf -= nr
for i := 0; i < d.nbuf; i++ {
d.buf[i] = d.buf[i+nr]
}
if d.err == nil {
d.err = err
}
return n, d.err
}
// NewDecoder constructs a new base62 stream decoder.
func NewDecoder(enc *Encoding, r io.Reader) io.Reader {
return &decoder{enc: enc, r: r}
}
// DecodedLen returns the maximum length in bytes of the decoded data
// corresponding to n bytes of base62-encoded data.
func (enc *Encoding) DecodedLen(n int) int { return n / 4 * 3 }
+35
View File
@@ -0,0 +1,35 @@
// Package hello
// @Author: Wang Feng
// @Description: base62编解码。源代码:https://github.com/lytics/base62
// @Version: 0.1.0
// @Date: 2022/7/14 22:20
// @Copyright: MIN-Group;国家重大科技基础设施——未来网络北大实验室;深圳市信息论与未来网络重点实验室
//
package utils
import (
"github.com/stretchr/testify/assert"
"testing"
)
func TestBase62(t *testing.T) {
urlVal := "http://www.lytics.io/?utm_content=content&utm_campaign=campaign"
encodedValue := Base62StdEncoding.EncodeToString([]byte(urlVal))
assert.NotEqual(t, "", encodedValue)
rtVal, err := Base62StdEncoding.DecodeString(encodedValue)
assert.Equal(t, nil, err)
assert.Equal(t, string(rtVal), urlVal)
intVal := "999"
encodedValue = Base62StdEncoding.EncodeToString([]byte(intVal))
assert.Equal(t, "OTk4", encodedValue, "%v", encodedValue)
rtVal, err = Base62StdEncoding.DecodeString(encodedValue)
assert.Equal(t, nil, err)
assert.Equal(t, string(rtVal), intVal)
urlVal = `http://our-uploads.s3.amazonaws.com/file-export/stuff-1427217700-12.csv?AWSAccessKeyId=AAAAIIG7MAQRTHTD7CLP&Expires=1427304113&Signature=VQsRAhgamiw1RVtbrCXOsMu%2BgFo`
encodedValue = Base62StdEncoding.EncodeToString([]byte(urlVal))
// We are just ensuring it doesn't panic
assert.True(t, len(encodedValue) > 0)
}