network

README

network

TCPServer

通过network.NewTCPServer创建TCPServer。

TCPServer使用ListenAndServe开启监听，接受新建立的连接并处理。

server := network.NewTCPServer("localhost:8000")
err := server.ListenAndServe(handler, codec)
if err != network.ErrServerClosed {
    // handle error
}

handler和codec实现连接事件处理和数据读写处理的接口，可以通过接口来实现自定义连接事件和数据读写处理。

TCPServer会对每个TCPConnection使用单独的协程来处理连接事件。

Error Handling

当ListenAndServe遇到不可恢复的错误时会将错误返回。

调用Close后，ListenAndServe会返回network.ErrClientClosed。

TCPClient

通过network.NewTCPClient创建TCPClient。

TCPClient使用DialAndServe主动建立连接并处理。

client := network.NewTCPClient("localhost:8000")
err := client.DialAndServe(handler, codec)
if err != network.ErrClientClosed {
    // handle error
}

handler和codec实现连接事件处理和数据读写处理的接口，可以通过接口来实现自定义连接事件和数据读写处理。

EnableRetry和DisableRetry用于启用和禁用当连接建立遇到错误或者失败时重试。

Error Handling

当重试未启用，DialAndServe遇到不可恢复的错误时会将错误返回。

调用Close后，DialAndServe会返回network.ErrClientClosed。

Codec接口

通过实现Read和Write接口来实现conn的数据读写处理方法。

type Codec interface {
    Read(r io.Reader) ([]byte, error)
    Write(w io.Writer, b []byte) error
}

数据处理接口：

Read方法实现如何从conn读取数据，返回读取的数据和是否遇到错误 Write方法实现如何将原始数据b写入conn，返回写入过程是否遇到错误

如果TCPServer或TCPClient的codec参数被设置为nil，则会调用network.DefaultCodec。

TCPHandler接口

通过实现TCPHandler来实现conn连接事件处理方法。

type TCPHandler interface {
    Connect(conn *TCPConnection, connected bool)
    Receive(conn *TCPConnection, buf []byte)
}

连接处理接口：

• Connect为连接事件，connected表示事件是建立或断开
• Receive为接收事件，buf表示接收到的内容（由Codec定义）
• conn是与事件相关的连接

如果TCPServer或TCPClient的handler参数被设置为nil，则会使用network.DefaultTCPHandler。

network.DefaultTCPHandler会忽略所有的连接事件。

Close & Graceful Shutdown

Close可以主动关闭连接，同时会直接丢弃队列中未发送的数据和丢弃接收缓冲区未读取的数据。

Shutdown()将设置关闭状态，并在给定的超时时间后（默认3秒）完全关闭socket，当发送协程发现关闭状态被设置，则会在缓冲队列所有数据都写入socket后调用conn.CloseWrite()通知对端已经写入完毕，此时等待对端主动关闭

或者，直接调用CloseWithTimeout(timeout time.Duration)例如：

conn.CloseWithTimeout(time.Second * 3)

吞吐量测试

乒乓测试（单机）

9827028992 total bytes read
2399177 total messages read
4096 average message size
937 MiB/s throughput

详细代码见pingpong_test.go

Documentation

Index

• Variables
• type Codec
• type CodecReader
• type CodecWriter
• type TCPClient
  • func NewTCPClient(addr string) *TCPClient
  • func (this *TCPClient) Close()
  • func (this *TCPClient) DialAndServe(handler TCPHandler, codec Codec) error
  • func (this *TCPClient) DisableRetry()
  • func (this *TCPClient) EnableRetry()
  • func (this *TCPClient) GetConnection() *TCPConnection
• type TCPConnection
  • func (this *TCPConnection) Close()
  • func (this *TCPConnection) CloseWithTimeout(timeout time.Duration)
  • func (this *TCPConnection) LocalAddr() net.Addr
  • func (this *TCPConnection) RemoteAddr() net.Addr
  • func (this *TCPConnection) Send(b []byte) error
  • func (this *TCPConnection) SetNoDelay(noDelay bool) error
  • func (this *TCPConnection) SetPendingSend(pendingSend int)
  • func (this *TCPConnection) Shutdown()
• type TCPHandler
• type TCPServer
  • func NewTCPServer(addr string) *TCPServer
  • func (this *TCPServer) Close()
  • func (this *TCPServer) ListenAndServe(handler TCPHandler, codec Codec) error

Variables

var DefaultCodec *defaultCodec = &defaultCodec{}

var DefaultTCPHandler = &defaultTCPHandler{}

var (
	ErrClientClosed = errors.New("network: Client closed")
)

var (
	ErrConnectionPendingSendFull = errors.New("network: Connection pending send full")
)

var (
	ErrServerClosed = errors.New("network: Server closed")
)

Types

type Codec

type Codec interface {
	Read(r io.Reader) ([]byte, error)
	Write(w io.Writer, b []byte) error
}

type CodecReader

type CodecReader interface {
	Read(r io.Reader) ([]byte, error)
}

type CodecWriter

type CodecWriter interface {
	Write(w io.Writer, b []byte) error
}

type TCPClient

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

func NewTCPClient

func NewTCPClient(addr string) *TCPClient

func (*TCPClient) Close

func (this *TCPClient) Close()

func (*TCPClient) DialAndServe

func (this *TCPClient) DialAndServe(handler TCPHandler, codec Codec) error

func (*TCPClient) DisableRetry

func (this *TCPClient) DisableRetry()

func (*TCPClient) EnableRetry

func (this *TCPClient) EnableRetry()

func (*TCPClient) GetConnection

func (this *TCPClient) GetConnection() *TCPConnection

type TCPConnection

type TCPConnection struct {
	Userdata interface{}
	// contains filtered or unexported fields
}

func (*TCPConnection) Close

func (this *TCPConnection) Close()

func (*TCPConnection) CloseWithTimeout

func (this *TCPConnection) CloseWithTimeout(timeout time.Duration)

func (*TCPConnection) LocalAddr

func (this *TCPConnection) LocalAddr() net.Addr

func (*TCPConnection) RemoteAddr

func (this *TCPConnection) RemoteAddr() net.Addr

func (*TCPConnection) Send

func (this *TCPConnection) Send(b []byte) error

func (*TCPConnection) SetNoDelay

func (this *TCPConnection) SetNoDelay(noDelay bool) error

func (*TCPConnection) SetPendingSend

func (this *TCPConnection) SetPendingSend(pendingSend int)

func (*TCPConnection) Shutdown

func (this *TCPConnection) Shutdown()

type TCPHandler

type TCPHandler interface {
	Connect(conn *TCPConnection, connected bool)
	Receive(conn *TCPConnection, buf []byte)
}

type TCPServer

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

func NewTCPServer

func NewTCPServer(addr string) *TCPServer

func (*TCPServer) Close

func (this *TCPServer) Close()

func (*TCPServer) ListenAndServe

func (this *TCPServer) ListenAndServe(handler TCPHandler, codec Codec) error