b612/notify
1
0
Fork 0
Code Issues Pull Requests Projects Releases 7 Wiki Activity

update new function

Browse Source
This commit is contained in:
兔子 2020-07-17 09:54:37 +08:00
parent 1358f40836
commit f2c7a55eda
5 changed files with 274 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
client.go
View File

@ -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) {

9
client_test.go
View File

@ -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()

29
serialization.go Normal file
View File

@ -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
}

148
server.go
View File

@ -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
}
}

93
starnotify/define.go Normal file
View File

@ -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
}