monitor_status/vendor/github.com/wpajqz/linker/tcp.go

package linker

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"runtime"
	"time"

	"github.com/wpajqz/linker/utils/convert"
)

func (s *Server) handleTCPConnection(conn *net.TCPConn) error {
	ctx := &ContextTcp{common: common{Context: context.Background()}, Conn: conn}
	if s.constructHandler != nil {
		s.constructHandler.Handle(ctx)
	}

	defer func() {
		if s.destructHandler != nil {
			s.destructHandler.Handle(ctx)
		}

		conn.Close()
	}()

	if s.config.ReadBufferSize > 0 {
		conn.SetReadBuffer(s.config.ReadBufferSize)
	}

	if s.config.WriteBufferSize > 0 {
		conn.SetWriteBuffer(s.config.WriteBufferSize)
	}

	var (
		bType         = make([]byte, 4)
		bSequence     = make([]byte, 8)
		bHeaderLength = make([]byte, 4)
		bBodyLength   = make([]byte, 4)
		sequence      int64
		headerLength  uint32
		bodyLength    uint32
	)

	for {
		if s.config.Timeout != 0 {
			conn.SetDeadline(time.Now().Add(s.config.Timeout))
		}

		if _, err := io.ReadFull(conn, bType); err != nil {
			return err
		}

		if _, err := io.ReadFull(conn, bSequence); err != nil {
			return err
		}

		if _, err := io.ReadFull(conn, bHeaderLength); err != nil {
			return err
		}

		if _, err := io.ReadFull(conn, bBodyLength); err != nil {
			return err
		}

		sequence = convert.BytesToInt64(bSequence)
		headerLength = convert.BytesToUint32(bHeaderLength)
		bodyLength = convert.BytesToUint32(bBodyLength)
		pacLen := headerLength + bodyLength + uint32(20)

		if pacLen > s.config.MaxPayload {
			_, file, line, _ := runtime.Caller(1)
			return SystemError{time.Now(), file, line, "packet larger than MaxPayload"}
		}

		header := make([]byte, headerLength)
		if _, err := io.ReadFull(conn, header); err != nil {
			return err
		}

		body := make([]byte, bodyLength)
		if _, err := io.ReadFull(conn, body); err != nil {
			return err
		}

		rp, err := NewPacket(convert.BytesToUint32(bType), sequence, header, body, s.config.PluginForPacketReceiver)

		if err != nil {
			return err
		}

		ctx = NewContextTcp(ctx.Context, conn, rp.Operator, rp.Sequence, rp.Header, rp.Body, s.config)
		go s.handleTCPPacket(ctx, rp)
	}
}

func (s *Server) handleTCPPacket(ctx Context, rp Packet) {
	defer func() {
		if r := recover(); r != nil {
			if s.errorHandler != nil {
				buf := make([]byte, 1<<12)
				n := runtime.Stack(buf, false)
				s.errorHandler(errors.New(string(buf[:n])))
			}
		}
	}()

	if rp.Operator == OPERATOR_HEARTBEAT {
		if s.pingHandler != nil {
			s.pingHandler.Handle(ctx)
		}

		ctx.Success(nil)
	}

	handler, ok := s.router.handlerContainer[rp.Operator]
	if !ok {
		ctx.Error(StatusInternalServerError, "server don't register your request.")
	}

	if rm, ok := s.router.routerMiddleware[rp.Operator]; ok {
		for _, v := range rm {
			ctx = v.Handle(ctx)
		}
	}

	for _, v := range s.router.middleware {
		ctx = v.Handle(ctx)
		if tm, ok := v.(TerminateMiddleware); ok {
			tm.Terminate(ctx)
		}
	}

	handler.Handle(ctx)
	ctx.Success(nil) // If it don't call the function of Success or Error, deal it by default
}

// RunTCP 开始运行Tcp服务
func (s *Server) RunTCP(name, address string) error {
	tcpAddr, err := net.ResolveTCPAddr(name, address)
	if err != nil {
		return err
	}

	listener, err := net.ListenTCP(name, tcpAddr)
	if err != nil {
		return err
	}

	defer listener.Close()

	fmt.Printf("tcp server running on %s\n", address)
	for {
		conn, err := listener.AcceptTCP()
		if err != nil {
			continue
		}

		go s.handleTCPConnection(conn)
	}
}