longtcp

Documentation

Index

• Constants
• Variables
• func CleanGlobalSessionPool(ip string, port int32)
• type ClientSessionPool
  • func GetGlobalSessionPool(ip string, port int32, timeout int, callbackhandshake HandshadeData, ...) *ClientSessionPool
  • func (sp *ClientSessionPool) Clean(err error) error
  • func (sp *ClientSessionPool) GetSession() (*Session, error)
• type HandshadeData
• type LongTCPHead
• type Message
  • func (m *Message) Desc() string
• type MessageID
• type MessageResult
• type OnReceivedFunc
• type OnSendDoneFunc
• type Session
  • func NewSession(ip string, port int32, timeout int, handshakedata []byte, ...) *Session
  • func NewSessionWithConn(conn *net.TCPConn, callback OnReceivedFunc) *Session
  • func (s *Session) Clean(err error)
  • func (s *Session) Desc() string
  • func (s *Session) IP() string
  • func (s *Session) IsValid() bool
  • func (s *Session) Send(data [][]byte, waitresponse bool, waitsecs int32, f OnSendDoneFunc) *MessageResult
  • func (s *Session) SendWithID(id MessageID, data [][]byte, waitresponse bool) *MessageResult
  • func (s *Session) Size() int64
• type TCPClient
  • func NewTCPClient(timeout int) *TCPClient
  • func NewTCPClientWithConn(conn *net.TCPConn) *TCPClient
  • func (c *TCPClient) Close() error
  • func (c *TCPClient) Closed() bool
  • func (c *TCPClient) ConnDesc() string
  • func (c *TCPClient) Connect(server string) error
  • func (c *TCPClient) LocalPort() int32
  • func (c *TCPClient) ReadData(expectlen int) ([]byte, int, error)
  • func (c *TCPClient) ReadUntilEOF() ([]byte, int, error)
  • func (c *TCPClient) RemoteAddr() string
  • func (c *TCPClient) SendFile(infile string, compress protocol.CompressType) error
  • func (c *TCPClient) SendMessages(messages []protocol.Message) error
  • func (c *TCPClient) TryReadData(expectlen int) ([]byte, int, error)
  • func (c *TCPClient) WriteData(data []byte) error

Constants

const (
	// 二进制中命令唯一标识的长度
	UniqIDLength = 32
	// 二进制中记录数据长度的字段的长度
	DataLengthInBinary     = 16
	TotalLongTCPHeadLength = 48

	DefaultLongTCPTimeoutSeconds = 3600 * 24
	MinWaitSecs                  = 300
)

Variables

var (
	ErrorContextCanceled        = fmt.Errorf("session canceled by context")
	ErrorConnectionInvalid      = fmt.Errorf("connection is invalid")
	ErrorResponseLengthInvalid  = fmt.Errorf("response data length is invalid")
	ErrorAllConnectionInvalid   = fmt.Errorf("all connections are invalid")
	ErrorSessionPoolCleaned     = fmt.Errorf("session pool cleaned")
	ErrorLessThanLongTCPHeadLen = fmt.Errorf("data length is less than long tcp head length")
	ErrorLongTCPHeadLenInvalid  = fmt.Errorf("data length of long tcp head length is invalid")
	ErrorWaitTimeout            = fmt.Errorf("wait response timeout")
)

Functions

func CleanGlobalSessionPool

func CleanGlobalSessionPool(ip string, port int32)

用于清理相应的session pool

Types

type ClientSessionPool

type ClientSessionPool struct {
	// contains filtered or unexported fields
}

---------------------------------------------------- 用于客户端的session pool

func GetGlobalSessionPool

func GetGlobalSessionPool(ip string, port int32, timeout int, callbackhandshake HandshadeData, size int32, callback OnReceivedFunc) *ClientSessionPool

用于初始化并返回session pool

func (*ClientSessionPool) Clean

func (sp *ClientSessionPool) Clean(err error) error

func (*ClientSessionPool) GetSession

func (sp *ClientSessionPool) GetSession() (*Session, error)

获取可用session

type HandshadeData

type HandshadeData func() []byte

type LongTCPHead

type LongTCPHead struct {
	UniqID  MessageID // 固定长度，用于识别命令，在二进制中的长度固定为32位
	DataLen int       // 记录data的长度，在二进制中的长度固定为16位
}

type Message

type Message struct {
	TCPHead      *LongTCPHead
	Data         [][]byte
	WaitResponse bool // 发送成功后，是否还需要等待对方返回结果
	RetChan      chan *MessageResult

	// 等待时间，由客户端指定;如果 MaxWaitSecs <= 0 ，则无限等待
	WaitStart   time.Time
	MaxWaitSecs int32

	F OnSendDoneFunc
}

约束条件： 返回结果的 UniqID 需要保持不变，方便收到结果后，找到对应的chan

func (*Message) Desc

func (m *Message) Desc() string

type MessageID

type MessageID string

固定长度UniqIDLength，用于识别命令

type MessageResult

type MessageResult struct {
	TCPHead *LongTCPHead
	Data    []byte
	Err     error
}

type OnReceivedFunc

type OnReceivedFunc func(id MessageID, data []byte, s *Session) error

处理收到的消息，一般是流程是将data转成需要的格式，然后业务逻辑处理，处理完，再通过 Session发送回去

type OnSendDoneFunc

type OnSendDoneFunc func() error

发送完成后的回调

type Session

type Session struct {
	// contains filtered or unexported fields
}

func NewSession

func NewSession(ip string, port int32, timeout int, handshakedata []byte, callback OnReceivedFunc) *Session

client端创建session，需要指定目标server的ip和端口 handshakedata : 用于建立握手协议的数据，当前兼容需要这个，后续不考虑兼容性，可以传nil callback : 收到数据后的处理函数

func NewSessionWithConn

func NewSessionWithConn(conn *net.TCPConn, callback OnReceivedFunc) *Session

server端创建session

func (*Session) Clean

func (s *Session) Clean(err error)

清理资源，包括关闭连接，停止协程等

func (*Session) Desc

func (s *Session) Desc() string

func (*Session) IP

func (s *Session) IP() string

func (*Session) IsValid

func (s *Session) IsValid() bool

func (*Session) Send

func (s *Session) Send(
	data [][]byte,
	waitresponse bool,
	waitsecs int32,
	f OnSendDoneFunc) *MessageResult

session 内部将data封装为Message发送，并通过chan接收发送结果，Message的id需要内部生成 如果 waitresponse为true，则需要等待返回的结果

func (*Session) SendWithID

func (s *Session) SendWithID(id MessageID, data [][]byte, waitresponse bool) *MessageResult

session 内部将data封装为Message发送，并通过chan接收发送结果，这儿指定了id，无需自动生成 如果 waitresponse为true，则需要等待返回的结果

func (*Session) Size

func (s *Session) Size() int64

返回当前未完成的任务数

type TCPClient

type TCPClient struct {
	// contains filtered or unexported fields
}

TCPClient wrapper net.TCPConn

func NewTCPClient

func NewTCPClient(timeout int) *TCPClient

NewTCPClient return new TCPClient

func NewTCPClientWithConn

func NewTCPClientWithConn(conn *net.TCPConn) *TCPClient

NewTCPClientWithConn return new TCPClient with specified conn

func (*TCPClient) Close

func (c *TCPClient) Close() error

Close close conn

func (*TCPClient) Closed

func (c *TCPClient) Closed() bool

Closed check if the TCP connection is closed

func (*TCPClient) ConnDesc

func (c *TCPClient) ConnDesc() string

ConnDesc return desc of conn

func (*TCPClient) Connect

func (c *TCPClient) Connect(server string) error

Connect connect to server

func (*TCPClient) LocalPort

func (c *TCPClient) LocalPort() int32

func (*TCPClient) ReadData

func (c *TCPClient) ReadData(expectlen int) ([]byte, int, error)

ReadData read data

func (*TCPClient) ReadUntilEOF

func (c *TCPClient) ReadUntilEOF() ([]byte, int, error)

ReadUntilEOF read data until EOF

func (*TCPClient) RemoteAddr

func (c *TCPClient) RemoteAddr() string

RemoteAddr return RemoteAddr

func (*TCPClient) SendFile

func (c *TCPClient) SendFile(infile string, compress protocol.CompressType) error

SendFile send file

func (*TCPClient) SendMessages

func (c *TCPClient) SendMessages(messages []protocol.Message) error

func (*TCPClient) TryReadData

func (c *TCPClient) TryReadData(expectlen int) ([]byte, int, error)

TryReadData try read data, return immediately after received any data

func (*TCPClient) WriteData

func (c *TCPClient) WriteData(data []byte) error

WriteData write data