add new feature

client.go

@@ -9,12 +9,13 @@ import (
 	"strings"
 	"time"
 
-	"b612.me/starainrt"
+	"b612.me/starnet"
 )
 
 // StarNotifyC 为Client端
 type StarNotifyC struct {
 	Connc       net.Conn
+	dialTimeout time.Duration
 	clientSign  map[string]chan string
 	// FuncLists 当不使用channel时，使用此记录调用函数
 	FuncLists   map[string]func(CMsg)
@@ -27,7 +28,7 @@ type StarNotifyC struct {
 	UseChannel bool
 	isUDP      bool
 	// Queue 是用来处理收发信息的简单消息队列
-	Queue *starainrt.StarQueue
+	Queue *starnet.StarQueue
 	// Online 当前链接是否处于活跃状态
 	Online   bool
 	lockPool map[string]CMsg
@@ -43,9 +44,12 @@ type CMsg struct {
 
 func (star *StarNotifyC) starinitc() {
 	star.stopSign, star.cancel = context.WithCancel(context.Background())
-	star.Queue = starainrt.NewQueue()
+	star.Queue = starnet.NewQueue()
+	star.Queue.EncodeFunc = encodeFunc
+	star.Queue.DecodeFunc = decodeFunc
+	star.Queue.Encode = true
 	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
@@ -71,6 +75,45 @@ func (star *StarNotifyC) store(key, value string) {
 	}
 	star.clientSign[key] <- value
 }
+func NewNotifyCWithTimeOut(netype, value string, timeout time.Duration) (*StarNotifyC, error) {
+	var err error
+	var star StarNotifyC
+	star.starinitc()
+	star.isUDP = false
+	if strings.Index(netype, "udp") >= 0 {
+		star.isUDP = true
+	}
+	star.Connc, err = net.DialTimeout(netype, value, timeout)
+	if err != nil {
+		return nil, err
+	}
+	star.dialTimeout = timeout
+	go star.cnotify()
+	go func() {
+		<-star.stopSign.Done()
+		star.Connc.Close()
+		star.Online = false
+		return
+	}()
+	go func() {
+		for {
+			buf := make([]byte, 8192)
+			n, err := star.Connc.Read(buf)
+			if n != 0 {
+				star.Queue.ParseMessage(buf[0:n], star.Connc)
+			}
+			if err != nil {
+				star.Connc.Close()
+				star.ClientStop()
+				//star, _ = NewNotifyC(netype, value)
+				star.Online = false
+				return
+			}
+		}
+	}()
+	star.Online = true
+	return &star, nil
+}
 
 // NewNotifyC 用于新建一个Client端进程
 func NewNotifyC(netype, value string) (*StarNotifyC, error) {
@@ -117,6 +160,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 +195,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) {

client_test.go

@@ -25,6 +25,7 @@ func Test_usechannel(t *testing.T) {
 		fmt.Println(err)
 		return
 	}
+	client.UseChannel = true
 	//time.Sleep(time.Second * 10)
 	client.Send("nihao")
 	client.SendValue("nihao", "lalala")
@@ -81,6 +82,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 +104,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()

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

server.go

@@ -10,28 +10,44 @@ import (
 	"strings"
 	"time"
 
-	"b612.me/starainrt"
+	"b612.me/starcrypto"
+
+	"b612.me/starnet"
 )
 
-var builder *starainrt.StarQueue
+var builder *starnet.StarQueue
+var aesKey = []byte{0x19, 0x96, 0x11, 0x27, 228, 187, 187, 231, 142, 137, 230, 179, 189, 229, 184, 133}
+
+func encodeFunc(data []byte) []byte {
+	return starcrypto.AesEncryptCFB(data, aesKey)
+}
+
+func decodeFunc(data []byte) []byte {
+	return starcrypto.AesDecryptCFB(data, aesKey)
+}
 
 func init() {
-	builder = starainrt.NewQueue()
+	builder = starnet.NewQueue()
+	builder.EncodeFunc = encodeFunc
+	builder.DecodeFunc = decodeFunc
+	builder.Encode = true
 }
 
 // StarNotifyS 为Server端
 type StarNotifyS struct {
 	// Queue 是用来处理收发信息的简单消息队列
-	Queue *starainrt.StarQueue
+	Queue *starnet.StarQueue
 	// 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
@@ -57,20 +73,58 @@ type SMsg struct {
 	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 +136,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 +166,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
@@ -135,10 +213,14 @@ func (star *StarNotifyS) SendWait(source SMsg, key, value string, tmout time.Dur
 
 func (star *StarNotifyS) starinits() {
 	star.stopSign, star.cancel = context.WithCancel(context.Background())
-	star.Queue = starainrt.NewQueue()
+	star.Queue = starnet.NewQueue()
+	star.Queue.EncodeFunc = encodeFunc
+	star.Queue.DecodeFunc = decodeFunc
+	star.Queue.Encode = true
 	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 +252,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 +271,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 +286,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 +300,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 +340,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 +360,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 +419,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
 			}
 		}

starnotify/define.go

@@ -0,0 +1,103 @@
+package starnotify
+
+import (
+	"errors"
+	"time"
+
+	"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 NewClientWithTimeout(key, netype, value string, timeout time.Duration) (*notify.StarNotifyC, error) {
+	client, err := notify.NewNotifyCWithTimeOut(netype, value, timeout)
+	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
+}