wsPool

README

wsPool

介绍

golang websocket 连接池

支持各种类型的数据交互

examples

func main() {
	flag.Parse()
	//初骀化连接池
	wsPool.InitWsPool(func(err interface{}) {
		//接收连接池中的运行时错误信息
		log.Panicln(err)
	})
	http.HandleFunc("/", serveHome)
	http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
		pcol := r.Header.Get("Sec-Websocket-Protocol")
		list:=strings.Split(pcol, "_")
		head := http.Header{}
		head.Add("Sec-Websocket-Protocol", pcol)

		//实例化连接对象
		client:=wsPool.NewClient(&wsPool.Config{
			Id:list[0], //连接标识
			Type:"ws", //连接类型
			Channel:list[1:], //指定频道
		})

		//开启连接
		client.OpenClient(w,r,head)

		//连接成功回调
		client.OnOpen(func() {
			log.Panicln("连接己开启"+client.Id)
		})

		//接收消息
		client.OnMessage(func(msg *wsPool.SendMsg) {
			log.Panicln(msg)
			if msg.ToClientId!="" {
				//发送消息给指定的ToClientID连接
				wsPool.Send(msg)
				//发送消息给当前连接对象
				client.Send(msg)
			}
			if len(msg.Channel)>0{
				//按频道广播，可指定多个频道[]string
				client.Broadcast(msg) //或者 wsPool.Broadcast(msg)
			}
			//或都全局广播，所有连接都进行发送
			wsPool.BroadcastAll(msg)

		})
		//连接断开回调
		client.OnClose(func() {
			log.Panicln("连接己经关闭"+client.Id)
		})
		client.OnError(func(err error) {
			log.Panicln("连接",client.Id,"错误信息：",err)
		})

	})
	err := http.ListenAndServe(*addr, nil)
	if err != nil {
		log.Fatal("ListenAndServe: ", err)
	}
}

作者很懒惰！！

其它看源码和例子，有些注释，很简单 ！

Documentation

Overview

Package grpool implements a goroutine reusable pool.

Index

• Variables
• func Broadcast(msg *Message, channel ...string) error
• func InitWsPool(errfun func(err interface{}))
• type Client
  • func GetClientById(id string) *Client
  • func NewClient(conf *Config) *Client
  • func (c *Client) Close()
  • func (c *Client) GetRuntimeInfo() *RuntimeInfo
  • func (c *Client) OnClose(h func())
  • func (c *Client) OnError(h func(err error))
  • func (c *Client) OnMessage(h func(msg []byte))
  • func (c *Client) OnMessageString(h func(msg string))
  • func (c *Client) OnOpen(h func())
  • func (c *Client) OpenClient(w http.ResponseWriter, r *http.Request, head http.Header)
  • func (c *Client) Send(msg *Message) error
• type Config
• type Message
• type Pool
  • func New(limit ...int) *Pool
  • func (p *Pool) Add(f func()) error
  • func (p *Pool) Cap() int
  • func (p *Pool) Close()
  • func (p *Pool) IsClosed() bool
  • func (p *Pool) Jobs() int
  • func (p *Pool) Size() int
• type RuntimeInfo
• type Server

Variables

var (
	//最大连接池缓冲处理连接对像管道长度
	Max_client_channel_len = 10240
	//最大全局广播缓冲处理管道长度
	Max_broadcastQueue_len = 4096
	//最大频道广播缓冲处理管道长度
	Max_chanBroadcastQueue_len = 4096

	//最大接收消息缓冲处理管道长度
	Max_recvCh_len = 10240
	//最大发送消息缓冲处理管道长度
	Max_sendCh_len = 10240
)

Functions

func Broadcast

func Broadcast(msg *Message, channel ...string) error

func InitWsPool

func InitWsPool(errfun func(err interface{}))

初始化执行连接池对象 参数为接收连接池中运行时的一些错误信息的回调方法

Types

type Client

type Client struct {
	CloseTime time.Time //连接断开的时间

	Id      string    //标识连接的名称
	IsClose chan bool //连接的状态。true为关闭
	// contains filtered or unexported fields
}

Client is a middleman between the websocket connection and the hub.

func GetClientById

func GetClientById(id string) *Client

通过id获取相应的连接对象

func NewClient

func NewClient(conf *Config) *Client

第一步，实例化连接对像

func (*Client) Close

func (c *Client) Close()

服务主动关闭连接

func (*Client) GetRuntimeInfo

func (c *Client) GetRuntimeInfo() *RuntimeInfo

获取连接对像运行过程中的信息

func (*Client) OnClose

func (c *Client) OnClose(h func())

监听连接对象的连接open成功的事件

func (*Client) OnError

func (c *Client) OnError(h func(err error))

监听连接对象的错误信息

func (*Client) OnMessage

func (c *Client) OnMessage(h func(msg []byte))

监听连接对象的连接open成功的事件

func (*Client) OnMessageString

func (c *Client) OnMessageString(h func(msg string))

监听连接对象的连接open成功的事件

func (*Client) OnOpen

func (c *Client) OnOpen(h func())

监听连接对象的连接open成功的事件

func (*Client) OpenClient

func (c *Client) OpenClient(w http.ResponseWriter, r *http.Request, head http.Header)

开启连接 serveWs handles websocket requests from the peer.

func (*Client) Send

func (c *Client) Send(msg *Message) error

单个连接发送消息

type Config

type Config struct {
	Id        string   //标识连接的名称
	Type      string   //连接类型或path
	Channel   []string //连接注册频道类型方便广播等操作。做为一个数组存储。因为一个连接可以属多个频道
	Goroutine int      //每个连接开启的go程数里 默认为10
}

连接参数结构体

type Message

type Message struct {
	MsgType int
	Message []byte
}

//接收消息结构 messageType=1 为string messageType=2 为[]byte

type Pool

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

Goroutine Pool

func New

func New(limit ...int) *Pool

New creates and returns a new goroutine pool object. The parameter <limit> is used to limit the max goroutine count, which is not limited in default.

func (*Pool) Add

func (p *Pool) Add(f func()) error

Add pushes a new job to the pool. The job will be executed asynchronously.

func (*Pool) Cap

func (p *Pool) Cap() int

Cap returns the capacity of the pool. This capacity is defined when pool is created. If it returns -1 means no limit.

func (*Pool) Close

func (p *Pool) Close()

Close closes the goroutine pool, which makes all goroutines exit.

func (*Pool) IsClosed

func (p *Pool) IsClosed() bool

IsClosed returns if pool is closed.

func (*Pool) Jobs

func (p *Pool) Jobs() int

Jobs returns current job count of the pool.

func (*Pool) Size

func (p *Pool) Size() int

Size returns current goroutine count of the pool.

type RuntimeInfo

type RuntimeInfo struct {
	Id              string //标识连接的名称
	Type            string //连接类型或path
	Ip              string
	Channel         []string  //连接注册频道类型方便广播等操作。做为一个数组存储。因为一个连接可以属多个频道
	OpenTime        time.Time //连接打开时间
	LastReceiveTime time.Time //最后一次接收到数据的时间
	LastSendTime    time.Time //最后一次发送数据的时间
}

type Server

type Server struct {
	ErrFun func(err interface{}) //用于接收ws连接池内代码运行时错误信息
	// contains filtered or unexported fields
}

连接池的结构体