sonet/internal/postal/logic/postal_cluster_server.go

package logic

import (
	"context"
	"errors"
	"fmt"
	"google.golang.org/grpc"
	"net"
	"sonet/api/gen/postal"
	"sonet/pkg/protocol/deliver"
	"sonet/pkg/protocol/session"
	"sonet/pkg/utils/logger"
	"sonet/pkg/utils/nets"
)

// PostalClusterPortOffset 相对与 postal grpc server 接口偏移量
const PostalClusterPortOffset = 1000

// PostalAddr2Cluster 根据 postal server 地址得到 postal cluster server 地址
func PostalAddr2Cluster(postalServerAddr string) (string, error) {
	addr, err := net.ResolveTCPAddr("tcp", postalServerAddr)
	if err != nil {
		return "", err
	}
	if addr.IP == nil {
		return fmt.Sprintf(":%d", addr.Port+PostalClusterPortOffset), nil
	}
	return fmt.Sprintf("%s:%d", addr.IP.String(), addr.Port+PostalClusterPortOffset), nil
}

var (
	methodRedirectDeliver      int32 = 10
	methodRedirectDeliverBatch int32 = 11
	methodRedirectDeliverGroup int32 = 12
)

// PostalClusterServer 用于postal server集群间, 消息转发通信
type PostalClusterServer struct {
	postal.UnimplementedPostalClusterServer
	postalServer postal.PostalServer // 当前节点 postal server
	postalPicker *deliver.PostalPicker
	sessionStore session.Store
}

func NewPostalClusterServer(postalServer postal.PostalServer, postalPicker *deliver.PostalPicker) *PostalClusterServer {
	return &PostalClusterServer{
		postalServer: postalServer,
		postalPicker: postalPicker,
	}
}

// Run 运行在同进程 postalServer 端口+1000
func (s *PostalClusterServer) Run(postalAddr string, opts ...grpc.ServerOption) (err error) {
	server := grpc.NewServer(opts...)
	postal.RegisterPostalClusterServer(server, s)
	// 解析端口
	address, err := PostalAddr2Cluster(postalAddr)
	if err != nil {
		return
	}
	listen, err := net.Listen("tcp", address)
	if err != nil {
		return
	}

	// run serve
	logger.Infof("%s grpc server running %s\n", postal.PostalCluster_ServiceDesc.ServiceName, listen.Addr().String())
	err = server.Serve(listen)
	return
}

func (s *PostalClusterServer) Redirect(ctx context.Context, req *postal.ReqRedirect) (*postal.ResDeliver, error) {
	req.Ttl -= 1
	switch req.RedirectMethod {
	case methodRedirectDeliver:
		return s.postalServer.Deliver(ctx, req.Deliver)
	case methodRedirectDeliverBatch:
		return s.postalServer.DeliverBatch(ctx, req.DeliverBatch)
	case methodRedirectDeliverGroup:
		return s.postalServer.DeliverGroup(ctx, req.DeliverGroup)
	}
	return nil, errors.New(fmt.Sprintf("not found redirect type %d", req.RedirectMethod))
}

func (s *PostalClusterServer) ReDeliver(ctx context.Context, req *postal.ReqReDeliver) (res *postal.ResDeliver, err error) {
	channel, ok := s.sessionStore.Load(req.Receiver)
	if ok {
		bytes, err := encodeDeliverMessage(req.Msg)
		if err != nil {
			return
		}
		err = channel.Conn.Write(bytes)
		if err != nil {
			return
		}
		res = &postal.ResDeliver{Ok: true}
		return
	}

	// 重新投递
	var next int
	next, req.Offset = convergeOffset(req.Offset)
	if next == 0 || req.Offset == 0 {
		// 重投次数结束
		return
	}
	node, _, err := s.postalPicker.PickOffsetNode(req.Receiver, next)
	if err != nil {
		return
	}
	addr, err := nets.Address2i64(node.GetAddr())
	if err != nil {
		return
	}
	for _, nodeI64 := range req.Nodes {
		if addr == nodeI64 {
			// 重复投递节点, 结束投递
			return
		}
	}

	req.Nodes = append(req.Nodes, addr)

	return
}

// convergeOffset 向0收敛offset
func convergeOffset(offset int32) (int, int32) {
	if offset < 0 {
		return -1, offset + 1
	}
	if offset > 0 {
		return 1, offset - 1
	}
	return 0, 0
}