From f30d42e130680ba9175ffda71ca551a399e73e7f Mon Sep 17 00:00:00 2001 From: 兔子 Date: Tue, 11 Feb 2020 10:50:11 +0800 Subject: [PATCH] update --- client.go | 35 ++++++++++++++++++----------------- server.go | 44 +++++++++++++++++++++----------------------- 2 files changed, 39 insertions(+), 40 deletions(-) diff --git a/client.go b/client.go index 0b65377..0f0f5ab 100644 --- a/client.go +++ b/client.go @@ -1,6 +1,7 @@ package notify import ( + "context" "net" "strings" "time" @@ -13,14 +14,14 @@ type StarNotifyC struct { Connc net.Conn clientSign map[string]chan string // FuncLists 当不使用channel时，使用此记录调用函数 - FuncLists map[string]func(CMsg) - clientStopSign chan int - defaultFunc func(CMsg) + FuncLists map[string]func(CMsg) + stopSign context.Context + cancel context.CancelFunc + defaultFunc func(CMsg) // Stop 停止信 号 Stop chan int // UseChannel 是否使用channel作为信息传递 UseChannel bool - notifychan chan int isUDP bool // Queue 是用来处理收发信息的简单消息队列 Queue *starainrt.StarQueue @@ -35,12 +36,14 @@ type CMsg struct { } func (star *StarNotifyC) starinitc() { + star.stopSign, star.cancel = context.WithCancel(context.Background()) star.Queue = starainrt.NewQueue() star.FuncLists = make(map[string]func(CMsg)) star.UseChannel = true - star.clientStopSign, star.notifychan, star.Stop = make(chan int, 1), make(chan int, 3), make(chan int, 5) + star.Stop = make(chan int, 5) star.clientSign = make(map[string]chan string) star.Online = false + star.Queue.RestoreDuration(time.Second * 2) } // Notify 用于获取一个通知 @@ -77,10 +80,8 @@ func NewNotifyC(netype, value string) (*StarNotifyC, error) { } go star.cnotify() go func() { - <-star.clientStopSign - star.notifychan <- 2 + <-star.stopSign.Done() star.Connc.Close() - star.Stop <- 0 star.Online = false return }() @@ -93,8 +94,7 @@ func NewNotifyC(netype, value string) (*StarNotifyC, error) { } if err != nil { star.Connc.Close() - star.Stop <- 1 - star.notifychan <- 3 + star.ClientStop() //star, _ = NewNotifyC(netype, value) star.Online = false return @@ -107,20 +107,20 @@ func NewNotifyC(netype, value string) (*StarNotifyC, error) { // Send 用于向Server端发送数据 func (star *StarNotifyC) Send(name string) error { - _, err := star.Connc.Write(star.Queue.BuildMessage(name)) + _, err := star.Connc.Write(star.Queue.BuildMessage([]byte(name))) return err } // SendValue 用于向Server端发送key-value类型数据 func (star *StarNotifyC) SendValue(name, value string) error { - _, err := star.Connc.Write(star.Queue.BuildMessage(name + "||" + value)) + _, err := star.Connc.Write(star.Queue.BuildMessage([]byte(name + "||" + value))) return err } func (star *StarNotifyC) cnotify() { for { select { - case <-star.notifychan: + case <-star.stopSign.Done(): return default: } @@ -129,12 +129,12 @@ func (star *StarNotifyC) cnotify() { time.Sleep(time.Millisecond * 20) continue } - if data.Msg == "b612ryzstop" { - star.clientStopSign <- 0 + if string(data.Msg) == "b612ryzstop" { + star.ClientStop() star.Online = false return } - strs := strings.SplitN(data.Msg, "||", 2) + strs := strings.SplitN(string(data.Msg), "||", 2) if len(strs) < 2 { continue } @@ -158,7 +158,8 @@ func (star *StarNotifyC) ClientStop() { if star.isUDP { star.Send("b612ryzstop") } - star.clientStopSign <- 1 + star.cancel() + star.Stop <- 1 } // SetNotify 用于设置关键词的调用函数 diff --git a/server.go b/server.go index b76fa19..abd5898 100644 --- a/server.go +++ b/server.go @@ -2,6 +2,7 @@ package notify import ( + "context" "net" "strings" "time" @@ -20,13 +21,13 @@ type StarNotifyS struct { // Queue 是用来处理收发信息的简单消息队列 Queue *starainrt.StarQueue // FuncLists 记录了被通知项所记录的函数 - FuncLists map[string]func(SMsg) string - defaultFunc func(SMsg) string - serverStopSign chan int - notifychan chan int - connPool map[string]net.Conn - udpPool map[string]*net.UDPAddr - isUDP bool + FuncLists map[string]func(SMsg) string + defaultFunc func(SMsg) string + stopSign context.Context + cancel context.CancelFunc + connPool map[string]net.Conn + udpPool map[string]*net.UDPAddr + isUDP bool // UDPConn UDP监听 UDPConn *net.UDPConn // Online 当前链接是否处于活跃状态 @@ -53,9 +54,9 @@ type SMsg struct { func (nmsg *SMsg) Reply(msg string) error { var err error if nmsg.uconn == nil { - _, err = nmsg.Conn.Write(builder.BuildMessage(nmsg.Key + "||" + msg)) + _, err = nmsg.Conn.Write(builder.BuildMessage([]byte(nmsg.Key + "||" + msg))) } else { - _, err = nmsg.uconn.WriteToUDP(builder.BuildMessage(nmsg.Key+"||"+msg), nmsg.UDP) + _, err = nmsg.uconn.WriteToUDP(builder.BuildMessage([]byte(nmsg.Key+"||"+msg)), nmsg.UDP) } return err } @@ -64,20 +65,21 @@ func (nmsg *SMsg) Reply(msg string) error { func (nmsg *SMsg) Send(key, value string) error { var err error if nmsg.uconn == nil { - _, err = nmsg.Conn.Write(builder.BuildMessage(key + "||" + value)) + _, err = nmsg.Conn.Write(builder.BuildMessage([]byte(key + "||" + value))) } else { - _, err = nmsg.uconn.WriteToUDP(builder.BuildMessage(key+"||"+value), nmsg.UDP) + _, err = nmsg.uconn.WriteToUDP(builder.BuildMessage([]byte(key+"||"+value)), nmsg.UDP) } return err } func (star *StarNotifyS) starinits() { - star.serverStopSign, star.notifychan = make(chan int, 1), make(chan int, 5) + star.stopSign, star.cancel = context.WithCancel(context.Background()) star.Queue = starainrt.NewQueue() star.udpPool = make(map[string]*net.UDPAddr) star.FuncLists = make(map[string]func(SMsg) string) star.connPool = make(map[string]net.Conn) star.Online = false + star.Queue.RestoreDuration(time.Second * 2) } // NewNotifyS 开启一个新的Server端通知 @@ -102,11 +104,9 @@ func doudps(netype, value string) (*StarNotifyS, error) { } go star.notify() go func() { - <-star.serverStopSign - star.notifychan <- 1 - star.notifychan <- 2 + <-star.stopSign.Done() for k, v := range star.udpPool { - star.UDPConn.WriteToUDP(star.Queue.BuildMessage("b612ryzstop"), v) + star.UDPConn.WriteToUDP(star.Queue.BuildMessage([]byte("b612ryzstop")), v) delete(star.connPool, k) } star.UDPConn.Close() @@ -140,9 +140,7 @@ func notudps(netype, value string) (*StarNotifyS, error) { } go star.notify() go func() { - <-star.serverStopSign - star.notifychan <- 3 - star.notifychan <- 4 + <-star.stopSign.Done() for k, v := range star.connPool { v.Close() delete(star.connPool, k) @@ -156,7 +154,7 @@ func notudps(netype, value string) (*StarNotifyS, error) { conn, err := listener.Accept() if err != nil { select { - case <-star.notifychan: + case <-star.stopSign.Done(): listener.Close() return default: @@ -206,7 +204,7 @@ func (star *StarNotifyS) SetDefaultNotify(name string, data func(SMsg) string) { func (star *StarNotifyS) notify() { for { select { - case <-star.notifychan: + case <-star.stopSign.Done(): return default: } @@ -215,7 +213,7 @@ func (star *StarNotifyS) notify() { time.Sleep(time.Millisecond * 20) continue } - key, value := analyseData(data.Msg) + key, value := analyseData(string(data.Msg)) var rmsg SMsg if !star.isUDP { rmsg = SMsg{data.Conn.(net.Conn), key, value, nil, nil} @@ -256,5 +254,5 @@ func analyseData(msg string) (key, value string) { // ServerStop 用于终止Server端运行 func (star *StarNotifyS) ServerStop() { - star.serverStopSign <- 0 + star.cancel() }