From f2c7a55edadbf0ad8649826e31cab701df91f67e Mon Sep 17 00:00:00 2001 From: 兔子 Date: Fri, 17 Jul 2020 09:54:37 +0800 Subject: [PATCH] update new function --- client.go | 17 ++++- client_test.go | 9 +++ serialization.go | 29 +++++++++ server.go | 148 +++++++++++++++++++++++++++++++++++++------ starnotify/define.go | 93 +++++++++++++++++++++++++++ 5 files changed, 274 insertions(+), 22 deletions(-) create mode 100644 serialization.go create mode 100644 starnotify/define.go diff --git a/client.go b/client.go index eda1e56..05f3f3c 100644 --- a/client.go +++ b/client.go @@ -45,7 +45,7 @@ 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.UseChannel = false star.Stop = make(chan int, 5) star.clientSign = make(map[string]chan string) star.Online = false @@ -117,6 +117,14 @@ func (star *StarNotifyC) Send(name string) error { return star.SendValue(name, "") } +func (star *StarNotifyC) SendValueRaw(key string, msg interface{}) error { + encodeData, err := encode(msg) + if err != nil { + return err + } + return star.SendValue(key, string(encodeData)) +} + // SendValue 用于向Server端发送key-value类型数据 func (star *StarNotifyC) SendValue(name, value string) error { var err error @@ -144,6 +152,13 @@ func (star *StarNotifyC) trim(name string) string { } return string(key) } +func (star *StarNotifyC) SendValueWaitRaw(key string, msg interface{}, tmout time.Duration) (CMsg, error) { + encodeData, err := encode(msg) + if err != nil { + return CMsg{}, err + } + return star.SendValueWait(key, string(encodeData), tmout) +} // SendValueWait 用于向Server端发送key-value类型数据并等待结果返回，此结果不会通过标准返回流程处理 func (star *StarNotifyC) SendValueWait(name, value string, tmout time.Duration) (CMsg, error) { diff --git a/client_test.go b/client_test.go index b07d7ba..3cf5f2f 100644 --- a/client_test.go +++ b/client_test.go @@ -81,6 +81,9 @@ func Test_pipec(t *testing.T) { return } server.SetNotify("ni\\||hao", func(data SMsg) string { + fmt.Println("name-get", data.GetName()) + fmt.Println("name-set", data.SetName("iiiis")) + fmt.Println("name-get", data.GetName()) fmt.Println("server recv:", data.Key, data.Value, data.mode) if data.Value != "" { data.Reply("nba") @@ -100,6 +103,12 @@ func Test_pipec(t *testing.T) { return } fmt.Println(sa) + sa, err = client.SendValueWait("ni\\||hao", "lalasdeee", time.Second*10) + if err != nil { + fmt.Println(err) + return + } + fmt.Println(sa) fmt.Println("sukidesu") time.Sleep(time.Second * 3) server.ServerStop() diff --git a/serialization.go b/serialization.go new file mode 100644 index 0000000..9871e2d --- /dev/null +++ b/serialization.go @@ -0,0 +1,29 @@ +package notify + +import ( + "bytes" + "encoding/gob" +) + +func Register(data interface{}) { + gob.Register(data) +} + +func RegisterAll(data []interface{}) { + for _, v := range data { + gob.Register(v) + } +} +func encode(src interface{}) ([]byte, error) { + var buf bytes.Buffer + enc := gob.NewEncoder(&buf) + err := enc.Encode(&src) + return buf.Bytes(), err +} + +func Decode(src []byte) (interface{}, error) { + dec := gob.NewDecoder(bytes.NewReader(src)) + var dst interface{} + err := dec.Decode(&dst) + return dst, err +} diff --git a/server.go b/server.go index bc77729..adf754a 100644 --- a/server.go +++ b/server.go @@ -26,12 +26,14 @@ type StarNotifyS struct { // FuncLists 记录了被通知项所记录的函数 FuncLists map[string]func(SMsg) string defaultFunc func(SMsg) string - Connected func(SMsg) string + Connected func(SMsg) + nickName map[string]string stopSign context.Context cancel context.CancelFunc connPool map[string]net.Conn lockPool map[string]SMsg udpPool map[string]*net.UDPAddr + listener net.Listener isUDP bool // Stop 停止信 号 Stop chan int @@ -50,27 +52,65 @@ type StarNotifyS struct { // SMsg 指明当前服务端被通知的关键字 type SMsg struct { - Conn net.Conn - Key string - Value string - UDP *net.UDPAddr - uconn *net.UDPConn - mode string - wait chan int + Conn net.Conn + Key string + Value string + UDP *net.UDPAddr + uconn *net.UDPConn + mode string + wait chan int + nickName func(string, string) error + getName func(string) string +} + +func (star *StarNotifyS) getName(conn string) string { + for k, v := range star.nickName { + if v == conn { + return k + } + } + return "" } // GetConnPool 获取所有Client端信息 func (star *StarNotifyS) GetConnPool() []SMsg { var result []SMsg for _, v := range star.connPool { - result = append(result, SMsg{Conn: v, mode: "pa"}) + result = append(result, SMsg{Conn: v, mode: "pa", nickName: star.setNickName, getName: star.getName}) } for _, v := range star.udpPool { - result = append(result, SMsg{UDP: v, uconn: star.UDPConn, mode: "pa0"}) + result = append(result, SMsg{UDP: v, uconn: star.UDPConn, mode: "pa0", nickName: star.setNickName, getName: star.getName}) } return result } +// GetConnPool 获取所有Client端信息 +func (star *StarNotifyS) GetClient(name string) (SMsg, error) { + if str, ok := star.nickName[name]; ok { + if conn, ok := star.connPool[str]; ok { + return SMsg{Conn: conn, mode: "pa", nickName: star.setNickName, getName: star.getName}, nil + } + if conn, ok := star.udpPool[str]; ok { + return SMsg{UDP: conn, uconn: star.UDPConn, mode: "pa0", nickName: star.setNickName, getName: star.getName}, nil + } + } + return SMsg{}, errors.New("Not Found") +} + +func (nmsg *SMsg) GetName() string { + if nmsg.uconn != nil { + return nmsg.getName(nmsg.UDP.String()) + } + return nmsg.getName(fmt.Sprint(nmsg.Conn)) +} + +func (nmsg *SMsg) SetName(name string) error { + if nmsg.uconn != nil { + return nmsg.nickName(name, nmsg.UDP.String()) + } + return nmsg.nickName(name, fmt.Sprint(nmsg.Conn)) +} + func (nmsg *SMsg) addSlash(name string) string { var key []byte for _, v := range []byte(name) { @@ -82,6 +122,14 @@ func (nmsg *SMsg) addSlash(name string) string { return string(key) } +func (nmsg *SMsg) ReplyRaw(msg interface{}) error { + encodeData, err := encode(msg) + if err != nil { + return err + } + return nmsg.Reply(string(encodeData)) +} + // Reply 用于向client端回复数据 func (nmsg *SMsg) Reply(msg string) error { var err error @@ -104,6 +152,22 @@ func (nmsg *SMsg) Send(key, value string) error { return err } +func (nmsg *SMsg) SendRaw(key string, msg interface{}) error { + encodeData, err := encode(msg) + if err != nil { + return err + } + return nmsg.Send(key, string(encodeData)) +} + +func (star *StarNotifyS) SendWaitRaw(source SMsg, key string, msg interface{}, tmout time.Duration) (SMsg, error) { + encodeData, err := encode(msg) + if err != nil { + return SMsg{}, err + } + return star.SendWait(source, key, string(encodeData), tmout) +} + // SendWait 用于向client端发送key-value数据，并等待 func (star *StarNotifyS) SendWait(source SMsg, key, value string, tmout time.Duration) (SMsg, error) { var err error @@ -139,6 +203,7 @@ func (star *StarNotifyS) starinits() { star.udpPool = make(map[string]*net.UDPAddr) star.FuncLists = make(map[string]func(SMsg) string) star.connPool = make(map[string]net.Conn) + star.nickName = make(map[string]string) star.lockPool = make(map[string]SMsg) star.Stop = make(chan int, 5) star.Online = false @@ -170,7 +235,12 @@ func doudps(netype, value string) (*StarNotifyS, error) { <-star.stopSign.Done() for k, v := range star.udpPool { star.UDPConn.WriteToUDP(star.Queue.BuildMessage([]byte("b612ryzstop")), v) - delete(star.connPool, k) + delete(star.udpPool, k) + for k2, v2 := range star.nickName { + if v2 == k { + delete(star.nickName, k2) + } + } } star.UDPConn.Close() star.Online = false @@ -184,7 +254,7 @@ func doudps(netype, value string) (*StarNotifyS, error) { star.Queue.ParseMessage(buf[0:n], addr) if _, ok := star.udpPool[addr.String()]; !ok { if star.Connected != nil { - go star.Connected(SMsg{UDP: addr, uconn: star.UDPConn}) + go star.Connected(SMsg{UDP: addr, uconn: star.UDPConn, nickName: star.setNickName, getName: star.getName}) } } star.udpPool[addr.String()] = addr @@ -199,10 +269,11 @@ func doudps(netype, value string) (*StarNotifyS, error) { } func notudps(netype, value string) (*StarNotifyS, error) { + var err error var star StarNotifyS star.starinits() star.isUDP = false - listener, err := net.Listen(netype, value) + star.listener, err = net.Listen(netype, value) if err != nil { return nil, err } @@ -212,18 +283,23 @@ func notudps(netype, value string) (*StarNotifyS, error) { for k, v := range star.connPool { v.Close() delete(star.connPool, k) + for k2, v2 := range star.nickName { + if v2 == k { + delete(star.nickName, k2) + } + } } - listener.Close() + star.listener.Close() star.Online = false return }() go func() { for { - conn, err := listener.Accept() + conn, err := star.listener.Accept() if err != nil { select { case <-star.stopSign.Done(): - listener.Close() + star.listener.Close() return default: continue @@ -247,14 +323,19 @@ func notudps(netype, value string) (*StarNotifyS, error) { } if err != nil { conn.Close() - delete(star.connPool, conn.RemoteAddr().String()) + delete(star.connPool, fmt.Sprint(conn)) + for k, v := range star.nickName { + if v == fmt.Sprint(conn) { + delete(star.nickName, k) + } + } break } } }(conn) - star.connPool[conn.RemoteAddr().String()] = conn + star.connPool[fmt.Sprint(conn)] = conn if star.Connected != nil { - go star.Connected(SMsg{Conn: conn}) + go star.Connected(SMsg{Conn: conn, nickName: star.setNickName, getName: star.getName}) } } }() @@ -262,6 +343,26 @@ func notudps(netype, value string) (*StarNotifyS, error) { return &star, nil } +func (star *StarNotifyS) GetListenerInfo() net.Listener { + return star.listener +} + +// SetNotify 用于设置通知关键词的调用函数 +func (star *StarNotifyS) setNickName(name string, conn string) error { + if _, ok := star.connPool[conn]; !ok { + if _, ok := star.udpPool[conn]; !ok { + return errors.New("Conn Not Found") + } + } + for k, v := range star.nickName { + if v == conn { + delete(star.nickName, k) + } + } + star.nickName[name] = conn + return nil +} + // SetNotify 用于设置通知关键词的调用函数 func (star *StarNotifyS) SetNotify(name string, data func(SMsg) string) { star.FuncLists[name] = data @@ -301,11 +402,16 @@ func (star *StarNotifyS) notify() { mode, key, value := star.analyseData(string(data.Msg)) var rmsg SMsg if !star.isUDP { - rmsg = SMsg{data.Conn.(net.Conn), key, value, nil, nil, mode, nil} + rmsg = SMsg{data.Conn.(net.Conn), key, value, nil, nil, mode, nil, star.setNickName, star.getName} } else { - rmsg = SMsg{nil, key, value, data.Conn.(*net.UDPAddr), star.UDPConn, mode, nil} + rmsg = SMsg{nil, key, value, data.Conn.(*net.UDPAddr), star.UDPConn, mode, nil, star.setNickName, star.getName} if key == "b612ryzstop" { delete(star.udpPool, rmsg.UDP.String()) + for k, v := range star.nickName { + if v == rmsg.UDP.String() { + delete(star.nickName, k) + } + } continue } } diff --git a/starnotify/define.go b/starnotify/define.go new file mode 100644 index 0000000..050340b --- /dev/null +++ b/starnotify/define.go @@ -0,0 +1,93 @@ +package starnotify + +import ( + "errors" + + "b612.me/notify" +) + +var ( + starClient map[string]*notify.StarNotifyC + starServer map[string]*notify.StarNotifyS +) + +func init() { + starClient = make(map[string]*notify.StarNotifyC) + starServer = make(map[string]*notify.StarNotifyS) +} + +func NewClient(key, netype, value string) (*notify.StarNotifyC, error) { + client, err := notify.NewNotifyC(netype, value) + if err != nil { + return client, err + } + starClient[key] = client + return client, err +} + +func DeleteClient(key string) error { + client, ok := starClient[key] + if !ok { + return errors.New("Not Exists Yet!") + } + if client.Online { + client.ClientStop() + } + client = nil + delete(starClient, key) + return nil +} + +func NewServer(key, netype, value string) (*notify.StarNotifyS, error) { + server, err := notify.NewNotifyS(netype, value) + if err != nil { + return server, err + } + starServer[key] = server + return server, err +} + +func DeleteServer(key string) error { + server, ok := starServer[key] + if !ok { + return errors.New("Not Exists Yet!") + } + if server.Online { + server.ServerStop() + } + server = nil + delete(starServer, key) + return nil +} + +func S(key string) *notify.StarNotifyS { + server, ok := starServer[key] + if !ok { + return nil + } + return server +} + +func C(key string) *notify.StarNotifyC { + client, ok := starClient[key] + if !ok { + return nil + } + return client +} + +func Server(key string) (*notify.StarNotifyS, error) { + server, ok := starServer[key] + if !ok { + return nil, errors.New("Not Exists Yet") + } + return server, nil +} + +func Client(key string) (*notify.StarNotifyC, error) { + client, ok := starClient[key] + if !ok { + return nil, errors.New("Not Exists Yet") + } + return client, nil +}