end_time

pool/client.go

@@ -209,14 +209,14 @@ loop:
 				break loop
 			}
 
-			/*	//ToClientId与Channel不能同时存在！！！注意！！！！
-				if message.ToClientId!="" {
-					Send(message)
-				}
-				//ToClientId与Channel不能同时存在！！！注意！！！！
-				if message.Channel!="" {
-					Broadcast(message)
-				}*/
+			//ToClientId与Channel不能同时存在！！！注意！！！！
+			//if message.ToClientId != "" {
+			//	Send(message)
+			//}
+			//ToClientId与Channel不能同时存在！！！注意！！！！
+			//if message.Channel != "" {
+			//	Broadcast(message)
+			//}
 
 			//收到消息触发回调
 			//c.onMessage(data)
@@ -266,19 +266,19 @@ Loop:
 				c.onError(errors.New("连接ID：" + c.Id + "写消息进写入IO错误！连接中断" + err.Error()))
 				goto Loop1
 			}
-			/*// Add queued chat messages to the current websocket message.
-			n := len(c.sendCh)
-			if n > 0 {
-				for i := 0; i < n; i++ {
+			// Add queued chat messages to the current websocket message.
+			//n := len(c.sendCh)
+			//if n > 0 {
+			//	for i := 0; i < n; i++ {
+			//
+			//		_, err = w.Write(<-c.sendCh)
+			//		if err != nil {
+			//			c.onError(errors.New("连接ID：" + c.Id + "写上次连接未发送的消息消息进写入IO错误！连接中断" + err.Error()))
+			//			return
+			//		}
+			//	}
+			//}
 
-					_, err = w.Write(<-c.sendCh)
-					if err != nil {
-						c.onError(errors.New("连接ID：" + c.Id + "写上次连接未发送的消息消息进写入IO错误！连接中断" + err.Error()))
-						return
-					}
-				}
-			}
-			*/
 			//关闭写入io对象
 			if err := w.Close(); err != nil {
 				c.onError(errors.New("连接ID：" + c.Id + "关闭写入IO对象出错，连接中断" + err.Error()))

pool/gopool.go

@@ -7,27 +7,25 @@ import (
 
 // Goroutine Pool
 type Pool struct {
-	limit  int         // Max goroutine count limit.
-	count  int  // Current running goroutine count.
-	list   *list.List // Job list for asynchronous job adding purpose.
-	closed bool // Is pool closed or not.
+	limit     int         // Max goroutine count limit.
+	count     int         // Current running goroutine count.
+	list      *list.List  // Job list for asynchronous job adding purpose.
+	closed    bool        // Is pool closed or not.
 	wJobsChan chan func() //写工作方法
 	rJobsChan chan func() //读工作方法
 }
 
-
-
 // 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 New(limit ...int) *Pool {
 	p := &Pool{
-		limit:  -1,
-		count:  0,
-		list:   list.New(),
-		closed: false,
-		wJobsChan:make(chan func()),
-		rJobsChan:make(chan func()),
+		limit:     -1,
+		count:     0,
+		list:      list.New(),
+		closed:    false,
+		wJobsChan: make(chan func()),
+		rJobsChan: make(chan func()),
 	}
 	if len(limit) > 0 && limit[0] > 0 {
 		p.limit = limit[0]
@@ -38,31 +36,28 @@ func New(limit ...int) *Pool {
 	return p
 }
 
-/*
-// Default goroutine pool.
-var pool = New()
-// Add pushes a new job to the pool using default goroutine pool.
-// The job will be executed asynchronously.
-func Add(f func()) error {
-	return pool.Add(f)
-}
-
-// Size returns current goroutine count of default goroutine pool.
-func Size() int {
-	return pool.Size()
-}
-
-// Jobs returns current job count of default goroutine pool.
-func Jobs() int {
-	return pool.Jobs()
-}
-*/
-
-
-func (p *Pool) runWrite(){
-	for !p.closed  {
+//// Default goroutine pool.
+//var pool = New()
+//// Add pushes a new job to the pool using default goroutine pool.
+//// The job will be executed asynchronously.
+//func Add(f func()) error {
+//	return pool.Add(f)
+//}
+//
+//// Size returns current goroutine count of default goroutine pool.
+//func Size() int {
+//	return pool.Size()
+//}
+//
+//// Jobs returns current job count of default goroutine pool.
+//func Jobs() int {
+//	return pool.Jobs()
+//}
+
+func (p *Pool) runWrite() {
+	for !p.closed {
 		select {
-		case f,ok:=<-p.wJobsChan:
+		case f, ok := <-p.wJobsChan:
 			if !ok {
 				break
 			}
@@ -71,11 +66,11 @@ func (p *Pool) runWrite(){
 	}
 }
 
-func (p *Pool) runRead(){
-	for !p.closed  {
+func (p *Pool) runRead() {
+	for !p.closed {
 		if job := p.list.Back(); job != nil {
 			value := p.list.Remove(job)
-			p.rJobsChan<-value.(func())
+			p.rJobsChan <- value.(func())
 		} else {
 			return
 		}
@@ -85,17 +80,17 @@ func (p *Pool) runRead(){
 // Add pushes a new job to the pool.
 // The job will be executed asynchronously.
 func (p *Pool) Add(f func()) error {
-	for p.closed{
+	for p.closed {
 		return errors.New("pool closed")
 	}
-	p.wJobsChan<-f
+	p.wJobsChan <- f
 
 	var n int
-		n = p.count
+	n = p.count
 	if p.limit != -1 && n >= p.limit {
 		return nil
 	}
-	p.count=n+1
+	p.count = n + 1
 	p.fork()
 	return nil
 }
@@ -125,7 +120,7 @@ func (p *Pool) fork() {
 		}()
 		for !p.closed {
 			select {
-			case job,ok:=<-p.rJobsChan:
+			case job, ok := <-p.rJobsChan:
 				if !ok {
 					break
 				}
@@ -142,7 +137,7 @@ func (p *Pool) IsClosed() bool {
 
 // Close closes the goroutine pool, which makes all goroutines exit.
 func (p *Pool) Close() {
-	p.closed=true
+	p.closed = true
 	close(p.wJobsChan)
 	close(p.rJobsChan)
 }

pool/publicApi.go

@@ -434,6 +434,14 @@ func PublishData(mqData *mq.SetMsgReq) {
 	return
 }
 
+func SetEndTime(){
+	mqData := &mq.SetMsgReq{
+		ProcedureType:8,
+	}
+	PublishData(mqData)
+	return
+}
+
 // 五分钟内 用户未连接、则判定为用户离线、这时候把会话置为离线状态、如果中间顾客 连线了则return、 如果中间客服离线了、则return
 //func (c *Client)SetOffline(user *UserInfo, customerId string) {
 //

ws_server.go

@@ -34,6 +34,14 @@ func main() {
 		fmt.Println("wsPool.InitWsPool error-------------", err)
 	})
 
+	ticker := time.NewTicker(time.Minute)
+	go func() {
+		for range ticker.C {
+			pool.SetEndTime()
+		}
+	}()
+
+
 	mux := http.NewServeMux()
 	mux.HandleFunc("/", serveHome)
 
@@ -105,18 +113,18 @@ func ws(w http.ResponseWriter, r *http.Request) {
 			//	fmt.Println("client.Send(msg):", err.Error())
 			//}
 		}
-		/*if len(msg.Channel)>0{
-			//按频道广播，可指定多个频道[]string
-			err:=pool.Broadcast(msg) //或者 wsPool.Broadcast(msg)
-			if err!=nil {
-				fmt.Println("pool.Broadcast(msg)", err.Error())
-			}
-		}
-		//或都全局广播，所有连接都进行发送
-		err:=pool.BroadcastAll(msg)
-		if err!=nil {
-			fmt.Println("pool.BroadcastAll(msg)", err.Error())
-		}*/
+		//if len(msg.Channel)>0{
+		//	//按频道广播，可指定多个频道[]string
+		//	err:=pool.Broadcast(msg) //或者 wsPool.Broadcast(msg)
+		//	if err!=nil {
+		//		fmt.Println("pool.Broadcast(msg)", err.Error())
+		//	}
+		//}
+		////或都全局广播，所有连接都进行发送
+		//err:=pool.BroadcastAll(msg)
+		//if err!=nil {
+		//	fmt.Println("pool.BroadcastAll(msg)", err.Error())
+		//}
 
 	})