添加推送服务

Author: hero

go_push/gateway/push_job.go

@@ -1,31 +1,71 @@
 package gateway
 
+import "fmt"
+
 type PushJob struct {
-	Type   int
-	roomId string
-	info   []byte
+	Type     int
+	PushType int
+	roomId   int
+	info     string
 }
 
 type PushTask struct {
 	JobChan          []chan *PushJob
 	DistributionTask chan *PushJob
 }
 
-func NewPushTask() *PushTask {
+func NewPushTask(roomLen, workNum, taskNum int) *PushTask {
 	var p *PushTask
-	p.DistributionTask = make(chan *PushJob, 10)
-	p.JobChan = make([]chan *PushJob, len(roomTitle)) //一个房间对应一个job
-	for i, _ := range roomTitle {
-		go p.distributionTask(i) //分发任务
+	p.DistributionTask = make(chan *PushJob, taskNum)
+	p.JobChan = make([]chan *PushJob, roomLen)
+	for i := 0; i < roomLen; i++ {
+		//可以一个房间开多个pushWork(i)
+		go p.pushWork(i) //分发任务
+	}
+	for i := 0; i < workNum; i++ {
+		go p.distributionTask()
 	}
 	return p
 }
 
-func (p *PushTask) distributionTask(roomId int) {
+func (p *PushTask) Push(job *PushJob) {
+	p.DistributionTask <- job
+}
 
+func (p *PushTask) distributionTask() {
+	var (
+		pushJob *PushJob
+	)
+	for {
+		select {
+		case pushJob = <-p.DistributionTask:
+			// 分发
+			if pushJob.Type == 1 {
+				GetRoomManage().PushAll(&WSMessage{
+					Type: pushJob.PushType,
+					Data: pushJob.info,
+				})
+			} else if pushJob.Type == 2 {
+				p.JobChan[pushJob.roomId] <- pushJob
+			}
+		}
+	}
 }
 
 func (p *PushTask) pushWork(roomId int) {
-     var (
-	 )
+	var (
+		err error
+		job *PushJob
+	)
+	for {
+		select {
+		case job = <-p.JobChan[roomId]:
+			if err = GetRoomManage().PushRoom(roomId, &WSMessage{
+				Type: job.PushType,
+				Data: job.info,
+			}); err != nil {
+				fmt.Println(err)
+			}
+		}
+	}
 }