add debug mode

client.go

@@ -1,330 +1,618 @@
 package notify
 
 import (
+	"b612.me/starcrypto"
+	"b612.me/stario"
+	"b612.me/starnet"
 	"context"
 	"errors"
 	"fmt"
 	"math/rand"
 	"net"
-	"strings"
+	"os"
+	"sync"
+	"sync/atomic"
 	"time"
-
-	"b612.me/starnet"
 )
 
-// StarNotifyC 为Client端
+func init() {
-type StarNotifyC struct {
+	Register(TransferMsg{})
-	Connc       net.Conn
-	dialTimeout time.Duration
-	clientSign  map[string]chan string
-	// FuncLists 当不使用channel时，使用此记录调用函数
-	FuncLists   map[string]func(CMsg)
-	stopSign    context.Context
-	cancel      context.CancelFunc
-	defaultFunc func(CMsg)
-	// Stop 停止信 号
-	Stop chan int
-	// UseChannel 是否使用channel作为信息传递
-	UseChannel bool
-	isUDP      bool
-	// Queue 是用来处理收发信息的简单消息队列
-	Queue *starnet.StarQueue
-	// Online 当前链接是否处于活跃状态
-	Online   bool
-	lockPool map[string]CMsg
 }
 
-// CMsg 指明当前客户端被通知的关键字
+type ClientCommon struct {
-type CMsg struct {
+	alive                      atomic.Value
-	Key   string
+	status                     Status
-	Value string
+	byeFromServer              bool
-	mode  string
+	conn                       net.Conn
-	wait  chan int
+	mu                         sync.Mutex
+	msgID                      uint64
+	queue                      *starnet.StarQueue
+	stopFn                     context.CancelFunc
+	stopCtx                    context.Context
+	parallelNum                int
+	maxReadTimeout             time.Duration
+	maxWriteTimeout            time.Duration
+	keyExchangeFn              func(c Client) error
+	linkFns                    map[string]func(message *Message)
+	defaultFns                 func(message *Message)
+	msgEn                      func([]byte, []byte) []byte
+	msgDe                      func([]byte, []byte) []byte
+	noFinSyncMsgPool           sync.Map
+	handshakeRsaPubKey         []byte
+	SecretKey                  []byte
+	noFinSyncMsgMaxKeepSeconds int
+	lastHeartbeat              int64
+	heartbeatPeriod            time.Duration
+	wg                         stario.WaitGroup
+	netType                    NetType
+	showError                  bool
+	skipKeyExchange            bool
+	useHeartBeat               bool
+	sequenceDe                 func([]byte) (interface{}, error)
+	sequenceEn                 func(interface{}) ([]byte, error)
+	debugMode                  bool
 }
 
-func (star *StarNotifyC) starinitc() {
+func (c *ClientCommon) Connect(network string, addr string) error {
-	star.stopSign, star.cancel = context.WithCancel(context.Background())
+	if c.alive.Load().(bool) {
-	star.Queue = starnet.NewQueue()
+		return errors.New("client already run")
-	star.Queue.EncodeFunc = encodeFunc
-	star.Queue.DecodeFunc = decodeFunc
-	star.Queue.Encode = true
-	star.FuncLists = make(map[string]func(CMsg))
-	star.UseChannel = false
-	star.Stop = make(chan int, 5)
-	star.clientSign = make(map[string]chan string)
-	star.Online = false
-	star.lockPool = make(map[string]CMsg)
-	star.Queue.RestoreDuration(time.Second * 2)
-}
-
-// Notify 用于获取一个通知
-func (star *StarNotifyC) Notify(key string) chan string {
-	if _, ok := star.clientSign[key]; !ok {
-		ch := make(chan string, 20)
-		star.clientSign[key] = ch
 	}
-	return star.clientSign[key]
+	c.stopCtx, c.stopFn = context.WithCancel(context.Background())
-}
+	c.queue = starnet.NewQueueCtx(c.stopCtx, 4)
-
+	conn, err := net.Dial(network, addr)
-func (star *StarNotifyC) store(key, value string) {
-	if _, ok := star.clientSign[key]; !ok {
-		ch := make(chan string, 20)
-		ch <- value
-		star.clientSign[key] = ch
-		return
-	}
-	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) {
-	var err error
-	var star StarNotifyC
-	star.starinitc()
-	star.isUDP = false
-	if strings.Index(netype, "udp") >= 0 {
-		star.isUDP = true
-	}
-	star.Connc, err = net.Dial(netype, value)
-	if err != nil {
-		return nil, err
-	}
-	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
-}
-
-// Send 用于向Server端发送数据
-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))
+	c.alive.Store(true)
+	c.status.Alive = true
+	c.conn = conn
+	if c.useHeartBeat {
+		go c.Heartbeat()
+	}
+	return c.clientPostInit()
 }
 
-// SendValue 用于向Server端发送key-value类型数据
+func (c *ClientCommon) DebugMode(dmg bool) {
-func (star *StarNotifyC) SendValue(name, value string) error {
+	c.mu.Lock()
-	var err error
+	c.debugMode = dmg
-	var key []byte
+	c.mu.Unlock()
-	for _, v := range []byte(name) {
-		if v == byte(124) || v == byte(92) {
-			key = append(key, byte(92))
-		}
-		key = append(key, v)
-	}
-	_, err = star.Connc.Write(star.Queue.BuildMessage([]byte("pa" + "||" + string(key) + "||" + value)))
-	return err
 }
 
-func (star *StarNotifyC) trim(name string) string {
+func (c *ClientCommon) IsDebugMode() bool {
-	var slash bool = false
+	return c.debugMode
-	var key []byte
-	for _, v := range []byte(name) {
-		if v == byte(92) && !slash {
-			slash = true
-			continue
-		}
-		slash = false
-		key = append(key, v)
-	}
-	return string(key)
 }
-func (star *StarNotifyC) SendValueWaitRaw(key string, msg interface{}, tmout time.Duration) (CMsg, error) {
+
-	encodeData, err := encode(msg)
+func (c *ClientCommon) ConnectTimeout(network string, addr string, timeout time.Duration) error {
+	if c.alive.Load().(bool) {
+		return errors.New("client already run")
+	}
+	c.stopCtx, c.stopFn = context.WithCancel(context.Background())
+	c.queue = starnet.NewQueueCtx(c.stopCtx, 4)
+	conn, err := net.DialTimeout(network, addr, timeout)
 	if err != nil {
-		return CMsg{}, err
+		return err
 	}
-	return star.SendValueWait(key, string(encodeData), tmout)
+	c.alive.Store(true)
+	c.status.Alive = true
+	c.conn = conn
+	if c.useHeartBeat {
+		go c.Heartbeat()
+	}
+	return c.clientPostInit()
 }
 
-// SendValueWait 用于向Server端发送key-value类型数据并等待结果返回，此结果不会通过标准返回流程处理
+func (c *ClientCommon) monitorPool() {
-func (star *StarNotifyC) SendValueWait(name, value string, tmout time.Duration) (CMsg, error) {
-	var err error
-	var tmceed <-chan time.Time
-	if star.UseChannel {
-		return CMsg{}, errors.New("Do Not Use UseChannel Mode!")
-	}
-	rand.Seed(time.Now().UnixNano())
-	mode := "cr" + fmt.Sprintf("%05d", rand.Intn(99999))
-	var key []byte
-	for _, v := range []byte(name) {
-		if v == byte(124) || v == byte(92) {
-			key = append(key, byte(92))
-		}
-		key = append(key, v)
-	}
-	_, err = star.Connc.Write(star.Queue.BuildMessage([]byte(mode + "||" + string(key) + "||" + value)))
-	if err != nil {
-		return CMsg{}, err
-	}
-	if int64(tmout) > 0 {
-		tmceed = time.After(tmout)
-	}
-	var source CMsg
-	source.wait = make(chan int, 2)
-	star.lockPool[mode] = source
-	select {
-	case <-source.wait:
-		res := star.lockPool[mode]
-		delete(star.lockPool, mode)
-		return res, nil
-	case <-tmceed:
-		return CMsg{}, errors.New("Time Exceed")
-	}
-}
-
-// ReplyMsg 用于向Server端Reply信息
-func (star *StarNotifyC) ReplyMsg(data CMsg, name, value string) error {
-	var err error
-	var key []byte
-	for _, v := range []byte(name) {
-		if v == byte(124) || v == byte(92) {
-			key = append(key, byte(92))
-		}
-		key = append(key, v)
-	}
-	_, err = star.Connc.Write(star.Queue.BuildMessage([]byte(data.mode + "||" + string(key) + "||" + value)))
-	return err
-}
-
-func (star *StarNotifyC) cnotify() {
 	for {
 		select {
-		case <-star.stopSign.Done():
+		case <-c.stopCtx.Done():
+			c.noFinSyncMsgPool.Range(func(k, v interface{}) bool {
+				data := v.(WaitMsg)
+				close(data.Reply)
+				c.noFinSyncMsgPool.Delete(k)
+				return true
+			})
+			return
+		case <-time.After(time.Second * 30):
+		}
+		now := time.Now()
+		if c.noFinSyncMsgMaxKeepSeconds > 0 {
+			c.noFinSyncMsgPool.Range(func(k, v interface{}) bool {
+				data := v.(WaitMsg)
+				if data.Time.Add(time.Duration(c.noFinSyncMsgMaxKeepSeconds) * time.Second).Before(now) {
+					close(data.Reply)
+					c.noFinSyncMsgPool.Delete(k)
+				}
+				return true
+			})
+		}
+	}
+}
+
+func (c *ClientCommon) SkipExchangeKey() bool {
+	return c.skipKeyExchange
+}
+
+func (c *ClientCommon) SetSkipExchangeKey(val bool) {
+	c.skipKeyExchange = val
+}
+
+func (c *ClientCommon) clientPostInit() error {
+	go c.readMessage()
+	go c.loadMessage()
+	if !c.skipKeyExchange {
+		err := c.keyExchangeFn(c)
+		if err != nil {
+			c.alive.Store(false)
+			c.mu.Lock()
+			c.status = Status{
+				Alive:  false,
+				Reason: "key exchange failed",
+				Err:    err,
+			}
+			c.mu.Unlock()
+			c.stopFn()
+			return err
+		}
+	}
+	return nil
+}
+func NewClient() Client {
+	var client = ClientCommon{
+		maxReadTimeout:     0,
+		maxWriteTimeout:    0,
+		sequenceEn:         encode,
+		sequenceDe:         Decode,
+		keyExchangeFn:      aesRsaHello,
+		SecretKey:          defaultAesKey,
+		handshakeRsaPubKey: defaultRsaPubKey,
+		msgEn:              defaultMsgEn,
+		msgDe:              defaultMsgDe,
+	}
+	client.alive.Store(false)
+	//heartbeat should not controlable for user
+	client.useHeartBeat = true
+	client.heartbeatPeriod = time.Second * 20
+	client.linkFns = make(map[string]func(*Message))
+	client.defaultFns = func(message *Message) {
+		return
+	}
+	client.wg = stario.NewWaitGroup(0)
+	client.stopCtx, client.stopFn = context.WithCancel(context.Background())
+	return &client
+}
+
+func (c *ClientCommon) Heartbeat() {
+	failedCount := 0
+	for {
+		select {
+		case <-c.stopCtx.Done():
+			return
+		case <-time.After(c.heartbeatPeriod):
+		}
+		_, err := c.sendWait(TransferMsg{
+			ID:    10000,
+			Key:   "heartbeat",
+			Value: nil,
+			Type:  MSG_SYS_WAIT,
+		}, time.Second*5)
+		if err == nil {
+			c.lastHeartbeat = time.Now().Unix()
+			failedCount = 0
+		}
+		if c.debugMode {
+			fmt.Println("failed to recv heartbeat,timeout!")
+		}
+		failedCount++
+		if failedCount >= 3 {
+			if c.debugMode {
+				fmt.Println("heatbeat failed more than 3 times,stop client")
+			}
+			c.alive.Store(false)
+			c.mu.Lock()
+			c.status = Status{
+				Alive:  false,
+				Reason: "heartbeat failed more than 3 times",
+				Err:    errors.New("heartbeat failed more than 3 times"),
+			}
+			c.mu.Unlock()
+			c.stopFn()
+			return
+		}
+	}
+}
+
+func (c *ClientCommon) ShowError(std bool) {
+	c.mu.Lock()
+	c.showError = std
+	c.mu.Unlock()
+}
+
+func (c *ClientCommon) readMessage() {
+	for {
+		select {
+		case <-c.stopCtx.Done():
+			c.conn.Close()
 			return
 		default:
 		}
-		data, err := star.Queue.RestoreOne()
+		data := make([]byte, 8192)
-		if err != nil {
+		if c.maxReadTimeout.Seconds() != 0 {
-			time.Sleep(time.Millisecond * 20)
+			if err := c.conn.SetReadDeadline(time.Now().Add(c.maxReadTimeout)); err != nil {
-			continue
+				//TODO:ALERT
-		}
-		if string(data.Msg) == "b612ryzstop" {
-			star.ClientStop()
-			star.Online = false
-			return
-		}
-		strs := strings.SplitN(string(data.Msg), "||", 3)
-		if len(strs) < 3 {
-			continue
-		}
-		strs[1] = star.trim(strs[1])
-		if star.UseChannel {
-			go star.store(strs[1], strs[2])
-		} else {
-			mode, key, value := strs[0], strs[1], strs[2]
-			if mode[0:2] != "cr" {
-				if msg, ok := star.FuncLists[key]; ok {
-					go msg(CMsg{key, value, mode, nil})
-				} else {
-					if star.defaultFunc != nil {
-						go star.defaultFunc(CMsg{key, value, mode, nil})
-					}
-				}
-			} else {
-				if sa, ok := star.lockPool[mode]; ok {
-					sa.Key = key
-					sa.Value = value
-					sa.mode = mode
-					star.lockPool[mode] = sa
-					sa.wait <- 1
-				} else {
-					if msg, ok := star.FuncLists[key]; ok {
-						go msg(CMsg{key, value, mode, nil})
-					} else {
-						if star.defaultFunc != nil {
-							go star.defaultFunc(CMsg{key, value, mode, nil})
-						}
-					}
-				}
 			}
 		}
+		readNum, err := c.conn.Read(data)
+		if err == os.ErrDeadlineExceeded {
+			if readNum != 0 {
+				c.queue.ParseMessage(data[:readNum], "b612")
+			}
+			continue
+		}
+		if err != nil {
+			if c.showError || c.debugMode {
+				fmt.Println("client read error", err)
+			}
+			c.alive.Store(false)
+			c.mu.Lock()
+			c.status = Status{
+				Alive:  false,
+				Reason: "client read error",
+				Err:    err,
+			}
+			c.mu.Unlock()
+			c.stopFn()
+			continue
+		}
+		c.queue.ParseMessage(data[:readNum], "b612")
 	}
 }
 
-// ClientStop 终止client端运行
+func (c *ClientCommon) sayGoodBye() error {
-func (star *StarNotifyC) ClientStop() {
+	_, err := c.sendWait(TransferMsg{
-	if star.isUDP {
+		ID:    10010,
-		star.Send("b612ryzstop")
+		Key:   "bye",
+		Value: nil,
+		Type:  MSG_SYS_WAIT,
+	}, time.Second*3)
+	return err
+}
+
+func (c *ClientCommon) loadMessage() {
+	for {
+		select {
+		case <-c.stopCtx.Done():
+			//say goodbye
+			if !c.byeFromServer {
+				c.sayGoodBye()
+			}
+			c.conn.Close()
+			return
+		case data, ok := <-c.queue.RestoreChan():
+			if !ok {
+				continue
+			}
+			c.wg.Add(1)
+			go func(data starnet.MsgQueue) {
+				defer c.wg.Done()
+				//fmt.Println("c received:", float64(time.Now().UnixNano()-nowd)/1000000)
+				now := time.Now()
+				//transfer to Msg
+				msg, err := c.sequenceDe(c.msgDe(c.SecretKey, data.Msg))
+				if err != nil {
+					if c.showError || c.debugMode {
+						fmt.Println("client decode data error", err)
+					}
+					return
+				}
+				message := Message{
+					ServerConn:  c,
+					TransferMsg: msg.(TransferMsg),
+					NetType:     NET_CLIENT,
+				}
+				message.Time = now
+				c.dispatchMsg(message)
+			}(data)
+		}
 	}
-	star.cancel()
-	star.Stop <- 1
-	star.Stop <- 1
-	star.Stop <- 1
 }
 
-// SetNotify 用于设置关键词的调用函数
+func (c *ClientCommon) dispatchMsg(message Message) {
-func (star *StarNotifyC) SetNotify(name string, data func(CMsg)) {
+	switch message.TransferMsg.Type {
-	star.FuncLists[name] = data
+	case MSG_SYS_WAIT:
+		fallthrough
+	case MSG_SYS:
+		c.sysMsg(message)
+		return
+	case MSG_KEY_CHANGE:
+		fallthrough
+	case MSG_SYS_REPLY:
+		fallthrough
+	case MSG_SYNC_REPLY:
+		data, ok := c.noFinSyncMsgPool.Load(message.ID)
+		if ok {
+			wait := data.(WaitMsg)
+			wait.Reply <- message
+			c.noFinSyncMsgPool.Delete(message.ID)
+			return
+		}
+		//return
+		fallthrough
+	default:
+	}
+	callFn := func(fn func(*Message)) {
+		fn(&message)
+	}
+	fn, ok := c.linkFns[message.Key]
+	if ok {
+		callFn(fn)
+	}
+	if c.defaultFns != nil {
+		callFn(c.defaultFns)
+	}
 }
 
-// SetDefaultNotify 用于设置默认关键词的调用函数
+func (c *ClientCommon) sysMsg(message Message) {
-func (star *StarNotifyC) SetDefaultNotify(data func(CMsg)) {
+	switch message.Key {
-	star.defaultFunc = data
+	case "bye":
+		if message.TransferMsg.Type == MSG_SYS_WAIT {
+			//fmt.Println("recv stop signal from server")
+			c.byeFromServer = true
+			message.Reply(nil)
+		}
+		c.alive.Store(false)
+		c.mu.Lock()
+		c.status = Status{
+			Alive:  false,
+			Reason: "recv stop signal from server",
+			Err:    nil,
+		}
+		c.mu.Unlock()
+		c.stopFn()
+	}
+}
+
+func (c *ClientCommon) SetDefaultLink(fn func(message *Message)) {
+	c.defaultFns = fn
+}
+
+func (c *ClientCommon) SetLink(key string, fn func(*Message)) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.linkFns[key] = fn
+}
+
+func (c *ClientCommon) send(msg TransferMsg) (WaitMsg, error) {
+	var wait WaitMsg
+	if msg.Type != MSG_SYNC_REPLY && msg.Type != MSG_KEY_CHANGE && msg.Type != MSG_SYS_REPLY || msg.ID == 0 {
+		msg.ID = atomic.AddUint64(&c.msgID, 1)
+	}
+	data, err := c.sequenceEn(msg)
+	if err != nil {
+		return WaitMsg{}, err
+	}
+	data = c.msgEn(c.SecretKey, data)
+	data = c.queue.BuildMessage(data)
+	if c.maxWriteTimeout.Seconds() != 0 {
+		c.conn.SetWriteDeadline(time.Now().Add(c.maxWriteTimeout))
+	}
+	_, err = c.conn.Write(data)
+	if err == nil && (msg.Type == MSG_SYNC_ASK || msg.Type == MSG_KEY_CHANGE || msg.Type == MSG_SYS_WAIT) {
+		wait.Time = time.Now()
+		wait.TransferMsg = msg
+		wait.Reply = make(chan Message, 1)
+		c.noFinSyncMsgPool.Store(msg.ID, wait)
+	}
+	return wait, err
+}
+
+func (c *ClientCommon) Send(key string, value MsgVal) error {
+	_, err := c.send(TransferMsg{
+		Key:   key,
+		Value: value,
+		Type:  MSG_ASYNC,
+	})
+	return err
+}
+
+func (c *ClientCommon) sendWait(msg TransferMsg, timeout time.Duration) (Message, error) {
+	data, err := c.send(msg)
+	if err != nil {
+		return Message{}, err
+	}
+	if timeout.Seconds() == 0 {
+		msg, ok := <-data.Reply
+		if !ok {
+			return msg, os.ErrInvalid
+		}
+		return msg, nil
+	}
+	select {
+	case <-time.After(timeout):
+		close(data.Reply)
+		c.noFinSyncMsgPool.Delete(data.TransferMsg.ID)
+		return Message{}, os.ErrDeadlineExceeded
+	case <-c.stopCtx.Done():
+		return Message{}, errors.New("service shutdown")
+	case msg, ok := <-data.Reply:
+		if !ok {
+			return msg, os.ErrInvalid
+		}
+		return msg, nil
+	}
+}
+
+func (c *ClientCommon) sendCtx(msg TransferMsg, ctx context.Context) (Message, error) {
+	data, err := c.send(msg)
+	if err != nil {
+		return Message{}, err
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	select {
+	case <-ctx.Done():
+		close(data.Reply)
+		c.noFinSyncMsgPool.Delete(data.TransferMsg.ID)
+		return Message{}, os.ErrDeadlineExceeded
+	case <-c.stopCtx.Done():
+		return Message{}, errors.New("service shutdown")
+	case msg, ok := <-data.Reply:
+		if !ok {
+			return msg, os.ErrInvalid
+		}
+		return msg, nil
+	}
+}
+
+func (c *ClientCommon) SendObjCtx(ctx context.Context, key string, val interface{}) (Message, error) {
+	data, err := c.sequenceEn(val)
+	if err != nil {
+		return Message{}, err
+	}
+	return c.sendCtx(TransferMsg{
+		Key:   key,
+		Value: data,
+		Type:  MSG_SYNC_ASK,
+	}, ctx)
+}
+
+func (c *ClientCommon) SendObj(key string, val interface{}) error {
+	data, err := encode(val)
+	if err != nil {
+		return err
+	}
+	_, err = c.send(TransferMsg{
+		Key:   key,
+		Value: data,
+		Type:  MSG_ASYNC,
+	})
+	return err
+}
+
+func (c *ClientCommon) SendCtx(ctx context.Context, key string, value MsgVal) (Message, error) {
+	return c.sendCtx(TransferMsg{
+		Key:   key,
+		Value: value,
+		Type:  MSG_SYNC_ASK,
+	}, ctx)
+}
+
+func (c *ClientCommon) SendWait(key string, value MsgVal, timeout time.Duration) (Message, error) {
+	return c.sendWait(TransferMsg{
+		Key:   key,
+		Value: value,
+		Type:  MSG_SYNC_ASK,
+	}, timeout)
+}
+
+func (c *ClientCommon) SendWaitObj(key string, value interface{}, timeout time.Duration) (Message, error) {
+	data, err := c.sequenceEn(value)
+	if err != nil {
+		return Message{}, err
+	}
+	return c.SendWait(key, data, timeout)
+}
+
+func (c *ClientCommon) Reply(m Message, value MsgVal) error {
+	return m.Reply(value)
+}
+
+func (c *ClientCommon) ExchangeKey(newKey []byte) error {
+	pubKey, err := starcrypto.DecodePublicKey(c.handshakeRsaPubKey)
+	if err != nil {
+		return err
+	}
+	newSendKey, err := starcrypto.RSAEncrypt(pubKey, newKey)
+	if err != nil {
+		return err
+	}
+	data, err := c.sendWait(TransferMsg{
+		ID:    19961127,
+		Key:   "sirius",
+		Value: newSendKey,
+		Type:  MSG_KEY_CHANGE,
+	}, time.Second*10)
+	if err != nil {
+		return err
+	}
+	if string(data.Value) != "success" {
+		return errors.New("cannot exchange new aes-key")
+	}
+	c.SecretKey = newKey
+	time.Sleep(time.Millisecond * 100)
+	return nil
+}
+
+func aesRsaHello(c Client) error {
+	newAesKey := []byte(fmt.Sprintf("%d%d%d%s", time.Now().UnixNano(), rand.Int63(), rand.Int63(), "b612.me"))
+	newAesKey = []byte(starcrypto.Md5Str(newAesKey))
+	return c.ExchangeKey(newAesKey)
+}
+
+func (c *ClientCommon) GetMsgEn() func([]byte, []byte) []byte {
+	return c.msgEn
+}
+func (c *ClientCommon) SetMsgEn(fn func([]byte, []byte) []byte) {
+	c.msgEn = fn
+}
+func (c *ClientCommon) GetMsgDe() func([]byte, []byte) []byte {
+	return c.msgDe
+}
+func (c *ClientCommon) SetMsgDe(fn func([]byte, []byte) []byte) {
+	c.msgDe = fn
+}
+
+func (c *ClientCommon) HeartbeatPeroid() time.Duration {
+	return c.heartbeatPeriod
+}
+func (c *ClientCommon) SetHeartbeatPeroid(duration time.Duration) {
+	c.heartbeatPeriod = duration
+}
+
+func (c *ClientCommon) GetSecretKey() []byte {
+	return c.SecretKey
+}
+func (c *ClientCommon) SetSecretKey(key []byte) {
+	c.SecretKey = key
+}
+func (c *ClientCommon) RsaPubKey() []byte {
+	return c.handshakeRsaPubKey
+}
+func (c *ClientCommon) SetRsaPubKey(key []byte) {
+	c.handshakeRsaPubKey = key
+}
+func (c *ClientCommon) Stop() error {
+	if !c.alive.Load().(bool) {
+		return nil
+	}
+	c.alive.Store(false)
+	c.mu.Lock()
+	c.status = Status{
+		Alive:  false,
+		Reason: "recv stop signal from user",
+		Err:    nil,
+	}
+	c.mu.Unlock()
+	c.stopFn()
+	return nil
+}
+func (c *ClientCommon) StopMonitorChan() <-chan struct{} {
+	return c.stopCtx.Done()
+}
+
+func (c *ClientCommon) Status() Status {
+	return c.status
+}
+
+func (c *ClientCommon) GetSequenceEn() func(interface{}) ([]byte, error) {
+	return c.sequenceEn
+}
+func (c *ClientCommon) SetSequenceEn(fn func(interface{}) ([]byte, error)) {
+	c.sequenceEn = fn
+}
+func (c *ClientCommon) GetSequenceDe() func([]byte) (interface{}, error) {
+	return c.sequenceDe
+}
+func (c *ClientCommon) SetSequenceDe(fn func([]byte) (interface{}, error)) {
+	c.sequenceDe = fn
 }

client_test.go

@@ -1,159 +0,0 @@
-package notify
-
-import (
-	"fmt"
-	"testing"
-	"time"
-)
-
-func Test_usechannel(t *testing.T) {
-	server, err := NewNotifyS("udp", "127.0.0.1:1926")
-	if err != nil {
-		fmt.Println(err)
-		return
-	}
-	server.SetNotify("nihao", func(data SMsg) string {
-		fmt.Println("server recv:", data.Key, data.Value)
-		if data.Value != "" {
-			data.Reply("nba")
-			return "nb"
-		}
-		return ""
-	})
-	client, err := NewNotifyC("udp", "127.0.0.1:1926")
-	if err != nil {
-		fmt.Println(err)
-		return
-	}
-	client.UseChannel = true
-	//time.Sleep(time.Second * 10)
-	client.Send("nihao")
-	client.SendValue("nihao", "lalala")
-	txt := <-client.Notify("nihao")
-	fmt.Println("client", txt)
-	txt = <-client.Notify("nihao")
-	fmt.Println("client", txt)
-	server.ServerStop()
-	<-client.Stop
-	client.ClientStop()
-	time.Sleep(time.Second * 3)
-}
-
-func Test_nochannel(t *testing.T) {
-	server, err := NewNotifyS("udp", "127.0.0.1:1926")
-	if err != nil {
-		fmt.Println(err)
-		return
-	}
-	server.SetNotify("nihao", func(data SMsg) string {
-		fmt.Println("server recv:", data.Key, data.Value)
-		if data.Value != "" {
-			data.Reply("nbaz")
-			return ""
-		}
-		return ""
-	})
-	client, err := NewNotifyC("udp", "127.0.0.1:1926")
-	if err != nil {
-		fmt.Println(err)
-		return
-	}
-	//time.Sleep(time.Second * 10)
-	client.UseChannel = false
-	client.SetNotify("nihao", func(data CMsg) {
-		fmt.Println("client recv:", data.Key, data.Value)
-		if data.Value != "" {
-			time.Sleep(time.Millisecond * 900)
-			client.SendValue("nihao", "dsb")
-		}
-	})
-	client.SendValue("nihao", "lalala")
-	time.Sleep(time.Second * 3)
-	server.ServerStop()
-	<-client.Stop
-	client.ClientStop()
-	time.Sleep(time.Second * 3)
-}
-
-func Test_pipec(t *testing.T) {
-	server, err := NewNotifyS("tcp", "127.0.0.1:1926")
-	if err != nil {
-		fmt.Println(err)
-		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")
-			return ""
-		}
-		return ""
-	})
-	client, err := NewNotifyC("tcp", "127.0.0.1:1926")
-	if err != nil {
-		fmt.Println(err)
-		return
-	}
-	client.UseChannel = false
-	sa, err := client.SendValueWait("ni\\||hao", "lalaeee", time.Second*10)
-	if err != nil {
-		fmt.Println(err)
-		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()
-	<-client.Stop
-	client.ClientStop()
-	time.Sleep(time.Second * 2)
-}
-
-func Test_pips(t *testing.T) {
-	var testmsg SMsg
-	server, err := NewNotifyS("udp", "127.0.0.1:1926")
-	if err != nil {
-		fmt.Println(err)
-		return
-	}
-	server.SetNotify("nihao", func(data SMsg) string {
-		fmt.Println("server recv:", data.Key, data.Value, data.mode)
-		testmsg = data
-		if data.Value != "" {
-			data.Reply("nbaz")
-			return ""
-		}
-		return ""
-	})
-	client, err := NewNotifyC("udp", "127.0.0.1:1926")
-	if err != nil {
-		fmt.Println(err)
-		return
-	}
-	//time.Sleep(time.Second * 10)
-	client.UseChannel = false
-	client.SetNotify("nihao", func(data CMsg) {
-		fmt.Println("client recv:", data.Key, data.Value, data.mode)
-		if data.mode != "pa" {
-			time.Sleep(time.Millisecond * 1200)
-			client.ReplyMsg(data, "nihao", "dsb")
-		}
-	})
-	client.SendValue("nihao", "lalala")
-	time.Sleep(time.Second * 3)
-	fmt.Println(server.SendWait(testmsg, "nihao", "wozuinb", time.Second*20))
-	fmt.Println("sakura")
-	server.ServerStop()
-	<-client.Stop
-	client.ClientStop()
-	time.Sleep(time.Second * 3)
-}

clienttype.go

@@ -0,0 +1,51 @@
+package notify
+
+import (
+	"context"
+	"time"
+)
+
+type Client interface {
+	SetDefaultLink(func(message *Message))
+	SetLink(string, func(*Message))
+	send(msg TransferMsg) (WaitMsg, error)
+	sendWait(msg TransferMsg, timeout time.Duration) (Message, error)
+	Send(key string, value MsgVal) error
+	SendWait(key string, value MsgVal, timeout time.Duration) (Message, error)
+	SendWaitObj(key string, value interface{}, timeout time.Duration) (Message, error)
+	SendCtx(ctx context.Context, key string, value MsgVal) (Message, error)
+	Reply(m Message, value MsgVal) error
+	ExchangeKey(newKey []byte) error
+	Connect(network string, addr string) error
+	ConnectTimeout(network string, addr string, timeout time.Duration) error
+	SkipExchangeKey() bool
+	SetSkipExchangeKey(bool)
+
+	GetMsgEn() func([]byte, []byte) []byte
+	SetMsgEn(func([]byte, []byte) []byte)
+	GetMsgDe() func([]byte, []byte) []byte
+	SetMsgDe(func([]byte, []byte) []byte)
+
+	Heartbeat()
+	HeartbeatPeroid() time.Duration
+	SetHeartbeatPeroid(duration time.Duration)
+
+	GetSecretKey() []byte
+	SetSecretKey(key []byte)
+	RsaPubKey() []byte
+	SetRsaPubKey([]byte)
+
+	Stop() error
+	StopMonitorChan() <-chan struct{}
+	Status() Status
+	ShowError(bool)
+	DebugMode(bool)
+	IsDebugMode() bool
+
+	GetSequenceEn() func(interface{}) ([]byte, error)
+	SetSequenceEn(func(interface{}) ([]byte, error))
+	GetSequenceDe() func([]byte) (interface{}, error)
+	SetSequenceDe(func([]byte) (interface{}, error))
+	SendObjCtx(ctx context.Context, key string, val interface{}) (Message, error)
+	SendObj(key string, val interface{}) error
+}

default.go

@@ -0,0 +1,82 @@
+package notify
+
+import (
+	"b612.me/starcrypto"
+)
+
+var defaultRsaKey = []byte(`-----BEGIN RSA PRIVATE KEY-----
+MIIJKAIBAAKCAgEAxmeMqr9yfJFKZn26oe/HvC7bZXNLC9Nk55AuTkb4XuIoqXDb
+AJD2Y/p167oJLKIqL3edcj7h+oTfn6s79vxT0ZCEf37ILU0G+scRzVwYHiLMwOUC
+bS2o4Xor3zqUi9f1piJBvoBNh8RKKtsmJW6VQZdiUGJHbgX4MdOdtf/6TvxZMwSX
+U+PRSCAjy04A31Zi7DEWUWJPyqmHeu++PxXU5lvoMdCGDqpcF2j2uO7oJJUww01M
+3F5FtTElMrK4/P9gD4kP7NiPhOfVPEfBsYT/DSSjvqNZJZuWnxu+cDxE7J/sBvdp
+eNRLhqzdmMYagZFuUmVrz8QmsD6jKHgydW+r7irllvb8WJPK/RIMif+4Rg7rDKFb
+j8+ZQ3HZ/gKELoRSyb3zL6RC2qlGLjC1tdeN7TNTinCv092y39T8jIARJ7tpfePh
+NBxsBdxfXbCAzHYZIHufI9Zlsc+felQwanlDhq+q8YLcnKHvNKYVyCf/upExpAiA
+rr88y/KbeKes0KorKkwMBnGUMTothWM25wHozcurixNvP4UMWX7LWD7vOZZuNDQN
+utZYeTwdsniI3mTO9vlPWEK8JTfxBU7x9SePUMJNDyjfDUJM8C2DOlyhGNPkgazO
+GdliH87tHkEy/7jJnGclgKmciiVPgwHfFx9GGoBHEfvmAoGGrk4qNbjm7JECAwEA
+AQKCAgBYzHe05ELFZfG6tYMWf08R9pbTbSqlfFOpIGrZNgJr1SUF0TDzq+3bCXpF
+qtn4VAw1en/JZkOV8Gp1+Bm6jWymWtwyg/fr7pG1I+vf0dwpgMHLg7P2UX1IjXmd
+S4a4oEuds69hJ+OLZFsdm0ATeM7ssGicOaBmqd1Pz7rCfnL1bxQtNVzVex1r/paG
+o77YNr3HoKCwhCPaPM4aQ7sOWSMUhwYBZabaYX0eLShf1O2pkexlPO+tobPpSLmx
+WzRYZ6QC0AGEq9hwT6KsfCFA5pmQtFllNY7suhpL1AsECLWAgoMNCyb1oW68NBpq
+CiBK5WBPGH2MW+pE74Pu1P0gen6kLGnApKQjprE1aGuR+xkZe3uEnXwSryU9TXki
+wINTEMsX8dkmofFqaJhUwSubrb+t7gvv9E9ZZe0X6UgKzAVVqvh4z1pP8VT+xHpu
+pW7SR8n9cFddaEPUijSb1rSpJrNzfJJ+G7yrB7Cw2kBgQ07vzD3z/3kA9cwFevLS
+mv3l3OQuB6y9c+AG3cX5WGAt/BVOLjimj9qJt+YglG0SwG31U0PUnnx6QVz/UtJm
+CbJQ2TpJd+mk0HyuMU+eycp7BWF3PMN+SE4QgKCKWnhsLeAd3gcvifsbLOYE1OPg
+wv1tqyJy0VsJiSn6Ub6Qq0kPLwCLlQTnLWk5mIhnRpHYufTSwQKCAQEA4gS4FKPU
+tAcQ82dEYW4OjGfhNWrjFpF+A8K5zufleQWcgzQ3fQho13zH0vZobukfkEVlVxla
+OIVk7ZgNA4mCSFrATjIx3RMqzrAUvTte0O4wkjYgCwVvTdS1W8nvRLKgugLygyoo
+r+MLW5IT3eNMK/2fZbftNlAkbc7NCo3c2tS6MXFgjx5JUuzChOY73Kp4p5KS38L5
+wRRiI8KTIKjBjMZ5q/l8VLKX89bKOCaWibmItoXY6QMbIjargb7YLp3X6uGEyGIu
+VhPbQ80/+OC2ZqIvDecp4PYnJNZFeqfjyfhJCNqDjBKYwIscBLMU/Wf9OY258OR4
+snQaerN1M0h9lQKCAQEA4LkZIRLLw+8bIVM+7VXxFwOAGy+MH35tvuNIToItAoUh
+zjL5LG34PjID8J0DPyP8VRVanak1EcxF0aTEkvnt2f2RAVsW89ytcn8Lybb12Ae8
+ia2ZWuIM+J40nuKOGPs3lJ9HqdPWmZYWsWKxFJmYBBnwD6CADYqhqambQn0HeaYl
+/WUD7blLYg+4Kk1mt9/hIw93jTWP/86O2H0ia+AhYPTqyvVXfIXKhat6NlOYksGf
+Hdv+aCC8Ukg6FyEgiNc/rFn0MWPnEX+cM1AwubviHIBhV8QWILLBTjupwsEBZVah
+60ftH+HRUCmEeOpI7jyzIlfEUNLoBHfswKMhMPtcDQKCAQEA0JFkQX+xn/PJW6PX
+AUWrXTvbIg0hw8i9DcFa76klJBnehWDhN5tUDE5Uo8PJOVgdTWgMjWSS0geezHX8
+xF/XfudoAIDnbMfsP9FTQhCQfaLf5XzW8vSv8pWwSiS9jJp+IUjo+8siwrR03aqe
+dKr0tr+ToS0qVG1+QGqO4gdpX/LgYxHp9ggPx9s94aAIa6hQMOrcaGqnSNqDedZr
+KL8x5LOewek3J32rJVP3Rfut/SfeFfjL4rKADoF+oPs4yUPVZSV4/+VCNyKZuyaj
+uwm6qFlPrLe9+J+OHbsxYG+fj9hzpRzoOZFLrppwX5HWc8XLcpnrlXVwP9VOPh5u
+r8VcRQKCAQAJFHGHfJLvH8Ig3pQ0UryjCWkrsAghXaJhjB1nzqqy514uTrDysp7N
+JIg0OKPg8TtI1MwMgsG6Ll7D0bx/k8mgfTZWr6+FuuznK2r2g4X7bJSZm4IOwgN0
+KDBIGy9SoxPj1Wu32O9a1U2lbS9qfao+wC2K9Bk4ctmFWW0Eiri6mZP/YQ1/lXUO
+SURPsUDtPQaDvCRAeGGRHG95H9U8NpoiqMKz4KXgSiecrwkJGOeZRml/c1wcKPZy
+/KgcNyJxZQEVnazYMgksE9Pj3uGZH5ZLQISuXyXlvFNDLfX2AIZl6dIxB371QtKK
+QqMvn4fC2IEEajdsbJkjVRUj03OL3xwhAoIBAAfMhDSvBbDkGTaXnNMjPPSbswqK
+qcSRhSG27mjs1dDNBKuFbz6TkIOp4nxjuS9Zp19fErXlAE9mF5yXSmuiAkZmWfhs
+HKpWIdjFJK1EqSfcINe2YuoyUIulz9oG7ObRHD4D8jSPjA8Ete+XsBHGyOtUl09u
+X4u9uClhqjK+r1Tno2vw5yF6ZxfQtdWuL4W0UL1S8E+VO7vjTjNOYvgjAIpAM/gW
+sqjA2Qw52UZqhhLXoTfRvtJilxlXXhIRJSsnUoGiYVCQ/upjqJCClEvJfIWdGY/U
+I2CbFrwJcNvOG1lUsSM55JUmbrSWVPfo7yq2k9GCuFxOy2n/SVlvlQUcNkA=
+-----END RSA PRIVATE KEY-----`)
+
+var defaultRsaPubKey = []byte(`-----BEGIN PUBLIC KEY-----
+MIICIjANBgkqhkiG9w0BAQEFAAOCAg8AMIICCgKCAgEAxmeMqr9yfJFKZn26oe/H
+vC7bZXNLC9Nk55AuTkb4XuIoqXDbAJD2Y/p167oJLKIqL3edcj7h+oTfn6s79vxT
+0ZCEf37ILU0G+scRzVwYHiLMwOUCbS2o4Xor3zqUi9f1piJBvoBNh8RKKtsmJW6V
+QZdiUGJHbgX4MdOdtf/6TvxZMwSXU+PRSCAjy04A31Zi7DEWUWJPyqmHeu++PxXU
+5lvoMdCGDqpcF2j2uO7oJJUww01M3F5FtTElMrK4/P9gD4kP7NiPhOfVPEfBsYT/
+DSSjvqNZJZuWnxu+cDxE7J/sBvdpeNRLhqzdmMYagZFuUmVrz8QmsD6jKHgydW+r
+7irllvb8WJPK/RIMif+4Rg7rDKFbj8+ZQ3HZ/gKELoRSyb3zL6RC2qlGLjC1tdeN
+7TNTinCv092y39T8jIARJ7tpfePhNBxsBdxfXbCAzHYZIHufI9Zlsc+felQwanlD
+hq+q8YLcnKHvNKYVyCf/upExpAiArr88y/KbeKes0KorKkwMBnGUMTothWM25wHo
+zcurixNvP4UMWX7LWD7vOZZuNDQNutZYeTwdsniI3mTO9vlPWEK8JTfxBU7x9SeP
+UMJNDyjfDUJM8C2DOlyhGNPkgazOGdliH87tHkEy/7jJnGclgKmciiVPgwHfFx9G
+GoBHEfvmAoGGrk4qNbjm7JECAwEAAQ==
+-----END PUBLIC KEY-----`)
+
+var defaultAesKey = []byte{0x19, 0x96, 0x11, 0x27, 228, 187, 187, 231, 142, 137, 230, 179, 189, 229, 184, 133}
+
+func defaultMsgEn(key []byte, d []byte) []byte {
+	return starcrypto.AesEncryptCFB(d, key)
+}
+
+func defaultMsgDe(key []byte, d []byte) []byte {
+	return starcrypto.AesDecryptCFB(d, key)
+}

go.mod

@@ -0,0 +1,9 @@
+module b612.me/notify
+
+go 1.16
+
+require (
+	b612.me/starcrypto v0.0.2
+	b612.me/stario v0.0.5
+	b612.me/starnet v0.1.3
+)

go.sum

@@ -0,0 +1,17 @@
+b612.me/starcrypto v0.0.2 h1:aRf1HcqK8GqHYxLAhWfFC4W/EqQLEFNEmxsBu3wG30o=
+b612.me/starcrypto v0.0.2/go.mod h1:hz0xRnfWNpYOlVrIPoGrQOWPibq4YiUZ7qN5tsQbzPo=
+b612.me/stario v0.0.5 h1:Q1OGF+8eOoK49zMzkyh80GWaMuknhey6+PWJJL9ZuNo=
+b612.me/stario v0.0.5/go.mod h1:or4ssWcxQSjMeu+hRKEgtp0X517b3zdlEOAms8Qscvw=
+b612.me/starnet v0.1.3 h1:UjY6M96gdPdJtxnQGzCttqSwFw93sDZSHiIGtdOlFfk=
+b612.me/starnet v0.1.3/go.mod h1:j/dd6BKwQK80O4gfbGYg2aYtPH76gSdgpuKboK/DwN4=
+golang.org/x/crypto v0.0.0-20220313003712-b769efc7c000 h1:SL+8VVnkqyshUSz5iNnXtrBQzvFF2SkROm6t5RczFAE=
+golang.org/x/crypto v0.0.0-20220313003712-b769efc7c000/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1 h1:SrN+KX8Art/Sf4HNj6Zcz06G7VEz+7w9tdXTPOZ7+l4=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1 h1:v+OssWQX+hTHEmOBgwxdZxK4zHq3yOs8F9J7mk0PY8E=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

msg.go

@@ -0,0 +1,502 @@
+package notify
+
+import (
+	"b612.me/starcrypto"
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"os"
+	"reflect"
+	"sync/atomic"
+	"time"
+)
+
+const (
+	MSG_SYS MessageType = iota
+	MSG_SYS_WAIT
+	MSG_SYS_REPLY
+	MSG_KEY_CHANGE
+	MSG_ASYNC
+	MSG_SYNC_ASK
+	MSG_SYNC_REPLY
+)
+
+type MessageType uint8
+
+type NetType uint8
+
+const (
+	NET_SERVER NetType = iota
+	NET_CLIENT
+)
+
+type MsgVal []byte
+type TransferMsg struct {
+	ID    uint64
+	Key   string
+	Value MsgVal
+	Type  MessageType
+}
+
+type Message struct {
+	NetType
+	ClientConn *ClientConn
+	ServerConn Client
+	TransferMsg
+	Time time.Time
+}
+
+type WaitMsg struct {
+	TransferMsg
+	Time  time.Time
+	Reply chan Message
+	//Ctx   context.Context
+}
+
+func (m *Message) Reply(value MsgVal) (err error) {
+	reply := TransferMsg{
+		ID:    m.ID,
+		Key:   m.Key,
+		Value: value,
+		Type:  m.Type,
+	}
+	if reply.Type == MSG_SYNC_ASK {
+		reply.Type = MSG_SYNC_REPLY
+	}
+	if reply.Type == MSG_SYS_WAIT {
+		reply.Type = MSG_SYS_REPLY
+	}
+	if m.NetType == NET_SERVER {
+		_, err = m.ClientConn.server.send(m.ClientConn, reply)
+	}
+	if m.NetType == NET_CLIENT {
+		_, err = m.ServerConn.send(reply)
+	}
+	return
+}
+
+func (m *Message) ReplyObj(value interface{}) (err error) {
+	data, err := encode(value)
+	if err != nil {
+		return err
+	}
+	return m.Reply(data)
+}
+
+type ClientConn struct {
+	alive           atomic.Value
+	status          Status
+	ClientID        string
+	ClientAddr      net.Addr
+	tuConn          net.Conn
+	server          Server
+	stopFn          context.CancelFunc
+	stopCtx         context.Context
+	maxReadTimeout  time.Duration
+	maxWriteTimeout time.Duration
+	msgEn           func([]byte, []byte) []byte
+	msgDe           func([]byte, []byte) []byte
+	handshakeRsaKey []byte
+	SecretKey       []byte
+	lastHeartBeat   int64
+}
+
+type Status struct {
+	Alive  bool
+	Reason string
+	Err    error
+}
+
+func (c *ClientConn) readTUMessage() {
+	for {
+		select {
+		case <-c.stopCtx.Done():
+			c.tuConn.Close()
+			c.server.removeClient(c)
+			return
+		default:
+		}
+		if c.maxReadTimeout.Seconds() > 0 {
+			c.tuConn.SetReadDeadline(time.Now().Add(c.maxReadTimeout))
+		}
+		data := make([]byte, 8192)
+		num, err := c.tuConn.Read(data)
+		if err == os.ErrDeadlineExceeded {
+			if num != 0 {
+				c.server.pushMessage(data[:num], c.ClientID)
+			}
+			continue
+		}
+		if err != nil {
+			//conn is broke
+			c.alive.Store(false)
+			c.status = Status{
+				Alive:  false,
+				Reason: "read error",
+				Err:    err,
+			}
+			c.stopFn()
+			continue
+		}
+		c.server.pushMessage(data[:num], c.ClientID)
+		//fmt.Println("finished:", float64(time.Now().UnixNano()-nowd)/1000000)
+	}
+}
+
+func (c *ClientConn) rsaDecode(message Message) {
+	privKey, err := starcrypto.DecodePrivateKey(c.handshakeRsaKey, "")
+	if err != nil {
+		fmt.Println(err)
+		message.Reply([]byte("failed"))
+		return
+	}
+	data, err := starcrypto.RSADecrypt(privKey, message.Value)
+	if err != nil {
+		fmt.Println(err)
+		message.Reply([]byte("failed"))
+		return
+	}
+	//fmt.Println("aes-key changed to", string(data))
+	message.Reply([]byte("success"))
+	c.SecretKey = data
+}
+
+func (c *ClientConn) sayGoodByeForTU() error {
+	_, err := c.server.sendWait(c, TransferMsg{
+		ID:    10010,
+		Key:   "bye",
+		Value: nil,
+		Type:  MSG_SYS_WAIT,
+	}, time.Second*3)
+	return err
+}
+
+func (c *ClientConn) GetSecretKey() []byte {
+	return c.SecretKey
+}
+func (c *ClientConn) SetSecretKey(key []byte) {
+	c.SecretKey = key
+}
+
+func (c *ClientConn) GetMsgEn() func([]byte, []byte) []byte {
+	return c.msgEn
+}
+func (c *ClientConn) SetMsgEn(fn func([]byte, []byte) []byte) {
+	c.msgEn = fn
+}
+func (c *ClientConn) GetMsgDe() func([]byte, []byte) []byte {
+	return c.msgDe
+}
+func (c *ClientConn) SetMsgDe(fn func([]byte, []byte) []byte) {
+	c.msgDe = fn
+}
+
+func (c *ClientConn) StopMonitorChan() <-chan struct{} {
+	return c.stopCtx.Done()
+}
+
+func (c *ClientConn) Status() Status {
+	return c.status
+}
+
+func (c *ClientConn) Server() Server {
+	return c.server
+}
+
+func (c *ClientConn) GetRemoteAddr() net.Addr {
+	return c.ClientAddr
+}
+
+func (m MsgVal) ToClearString() string {
+	return string(m)
+}
+
+func (m MsgVal) ToInterface() (interface{}, error) {
+	return Decode(m)
+}
+
+func (m MsgVal) MustToInterface() interface{} {
+	inf, err := m.ToInterface()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToString() (string, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return "", err
+	}
+	if data, ok := inf.(string); !ok {
+		return "", errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToString() string {
+	inf, err := m.ToString()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToInt32() (int32, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return 0, err
+	}
+	if data, ok := inf.(int32); !ok {
+		return 0, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToInt32() int32 {
+	inf, err := m.ToInt32()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToInt() (int, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return 0, err
+	}
+	if data, ok := inf.(int); !ok {
+		return 0, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToInt() int {
+	inf, err := m.ToInt()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToUint64() (uint64, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return 0, err
+	}
+	if data, ok := inf.(uint64); !ok {
+		return 0, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToUint64() uint64 {
+	inf, err := m.ToUint64()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToUint32() (uint32, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return 0, err
+	}
+	if data, ok := inf.(uint32); !ok {
+		return 0, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToUint32() uint32 {
+	inf, err := m.ToUint32()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToUint() (uint, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return 0, err
+	}
+	if data, ok := inf.(uint); !ok {
+		return 0, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToUint() uint {
+	inf, err := m.ToUint()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToBool() (bool, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return false, err
+	}
+	if data, ok := inf.(bool); !ok {
+		return false, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToBool() bool {
+	inf, err := m.ToBool()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToFloat64() (float64, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return 0, err
+	}
+	if data, ok := inf.(float64); !ok {
+		return 0, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToFloat64() float64 {
+	inf, err := m.ToFloat64()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+func (m MsgVal) ToFloat32() (float32, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return 0, err
+	}
+	if data, ok := inf.(float32); !ok {
+		return 0, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToFloat32() float32 {
+	inf, err := m.ToFloat32()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToSliceString() ([]string, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return []string{}, err
+	}
+	if data, ok := inf.([]string); !ok {
+		return []string{}, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToSliceString() []string {
+	inf, err := m.ToSliceString()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToSliceInt64() ([]int64, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return []int64{}, err
+	}
+	if data, ok := inf.([]int64); !ok {
+		return []int64{}, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToSliceInt64() []int64 {
+	inf, err := m.ToSliceInt64()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func (m MsgVal) ToSliceFloat64() ([]float64, error) {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return []float64{}, err
+	}
+	if data, ok := inf.([]float64); !ok {
+		return []float64{}, errors.New("source data not match target type")
+	} else {
+		return data, nil
+	}
+}
+
+func (m MsgVal) MustToSliceFloat64() []float64 {
+	inf, err := m.ToSliceFloat64()
+	if err != nil {
+		panic(err)
+	}
+	return inf
+}
+
+func ToMsgVal(val interface{}) (MsgVal, error) {
+	return Encode(val)
+}
+
+func MustToMsgVal(val interface{}) MsgVal {
+	d, err := ToMsgVal(val)
+	if err != nil {
+		panic(err)
+	}
+	return d
+}
+
+func (m MsgVal) Orm(stu interface{}) error {
+	inf, err := m.ToInterface()
+	if err != nil {
+		return err
+	}
+	t := reflect.TypeOf(stu)
+	if t.Kind() != reflect.Ptr {
+		return errors.New("interface not writable(pointer wanted)")
+	}
+	if !reflect.ValueOf(stu).Elem().CanSet() {
+		return errors.New("interface not writable")
+	}
+	it := reflect.TypeOf(inf)
+	if t.Elem().Kind() != it.Kind() {
+		return fmt.Errorf("interface{} kind is %v,not %v", t.Elem().Kind(), it.Kind())
+	}
+	if t.Elem().Name() != it.Name() {
+		return fmt.Errorf("interface{} name is %v,not %v", t.Elem().Name(), it.Name())
+	}
+	if t.Elem().String() != it.String() {
+		return fmt.Errorf("interface{} string is %v,not %v", t.Elem().String(), it.String())
+	}
+	reflect.ValueOf(stu).Elem().Set(reflect.ValueOf(inf))
+	return nil
+}

msg_test.go

@@ -0,0 +1,96 @@
+package notify
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+func TestMsgEnDeCode(t *testing.T) {
+	Register(HelloMessage{})
+	Register(Error{})
+	go ServerRun(time.Second * 30)
+	time.Sleep(time.Second * 2)
+	ClientRun(time.Second * 35)
+}
+
+type Error struct {
+	Msg string
+}
+
+func (e Error) Error() string {
+	return e.Msg
+}
+
+type WorldMessage struct {
+	Port   int
+	MyCode string
+	MyInfo []int
+	Err    error
+}
+
+type HelloMessage struct {
+	ID      string
+	MyMap   map[string]string
+	MySlice []int
+	World   WorldMessage
+}
+
+func ClientRun(stopTime time.Duration) {
+	c := NewClient()
+	err := c.Connect("tcp", "127.0.0.1:23456")
+	if err != nil {
+		panic(err)
+	}
+	c.SetLink("msg", func(msg *Message) {
+		var hi HelloMessage
+		err := msg.Value.Orm(&hi)
+		if err != nil {
+			panic(err)
+		}
+		fmt.Printf("recv info from server,struct detail is %+v\n", hi)
+	})
+	timer := time.NewTimer(stopTime)
+	for {
+		select {
+		case <-timer.C:
+			c.Stop()
+			return
+		case <-time.After(time.Second * 2):
+			fmt.Println("client msg sent", c.SendObj("msg", HelloMessage{
+				ID:      "client",
+				MyMap:   map[string]string{"hello": "world"},
+				MySlice: []int{int(time.Now().Unix())},
+				World: WorldMessage{
+					Port:   520,
+					MyCode: "b612",
+					MyInfo: []int{0, 1, 2, 3},
+					Err:    Error{Msg: "Hello World"},
+				},
+			}))
+		}
+
+	}
+}
+
+func ServerRun(stopTime time.Duration) {
+	s := NewServer()
+	err := s.Listen("tcp", "127.0.0.1:23456")
+	if err != nil {
+		panic(err)
+	}
+	s.SetLink("msg", func(msg *Message) {
+		var hi HelloMessage
+		err := msg.Value.Orm(&hi)
+		if err != nil {
+			panic(err)
+		}
+		fmt.Printf("recv info from client:%v,struct detail is %+v\n", msg.ClientConn.GetRemoteAddr(), hi)
+		hi.ID = "server"
+		hi.World.Port = 666
+		hi.MySlice = append(hi.MySlice, 1, 1, 2, 7)
+		msg.ReplyObj(hi)
+	})
+	<-time.After(stopTime)
+	s.Stop()
+}

serialization.go

@@ -21,6 +21,10 @@ func encode(src interface{}) ([]byte, error) {
 	return buf.Bytes(), err
 }
 
+func Encode(src interface{}) ([]byte, error) {
+	return encode(src)
+}
+
 func Decode(src []byte) (interface{}, error) {
 	dec := gob.NewDecoder(bytes.NewReader(src))
 	var dst interface{}

servertype.go

@@ -0,0 +1,49 @@
+package notify
+
+import (
+	"context"
+	"net"
+	"time"
+)
+
+type Server interface {
+	SetDefaultCommEncode(func([]byte, []byte) []byte)
+	SetDefaultCommDecode(func([]byte, []byte) []byte)
+	SetDefaultLink(func(message *Message))
+	SetLink(string, func(*Message))
+	send(c *ClientConn, msg TransferMsg) (WaitMsg, error)
+	sendWait(c *ClientConn, msg TransferMsg, timeout time.Duration) (Message, error)
+	SendObjCtx(ctx context.Context, c *ClientConn, key string, val interface{}) (Message, error)
+	SendObj(c *ClientConn, key string, val interface{}) error
+	Send(c *ClientConn, key string, value MsgVal) error
+	SendWait(c *ClientConn, key string, value MsgVal, timeout time.Duration) (Message, error)
+	SendWaitObj(c *ClientConn, key string, value interface{}, timeout time.Duration) (Message, error)
+	SendCtx(ctx context.Context, c *ClientConn, key string, value MsgVal) (Message, error)
+	Reply(m Message, value MsgVal) error
+	pushMessage([]byte, string)
+	removeClient(client *ClientConn)
+	Listen(network string, addr string) error
+	Stop() error
+	StopMonitorChan() <-chan struct{}
+	Status() Status
+
+	GetSecretKey() []byte
+	SetSecretKey(key []byte)
+	RsaPrivKey() []byte
+	SetRsaPrivKey([]byte)
+
+	GetClient(id string) *ClientConn
+	GetClientLists() []*ClientConn
+	GetClientAddrs() []net.Addr
+
+	GetSequenceEn() func(interface{}) ([]byte, error)
+	SetSequenceEn(func(interface{}) ([]byte, error))
+	GetSequenceDe() func([]byte) (interface{}, error)
+	SetSequenceDe(func([]byte) (interface{}, error))
+	ShowError(bool)
+	DebugMode(bool)
+	IsDebugMode() bool
+
+	HeartbeatTimeoutSec() int64
+	SetHeartbeatTimeoutSec(int64)
+}

starnotify/define.go

@@ -1,101 +1,107 @@
 package starnotify
 
 import (
-	"errors"
-	"time"
-
 	"b612.me/notify"
+	"errors"
+	"sync"
 )
 
 var (
-	starClient map[string]*notify.StarNotifyC
+	cmu        sync.RWMutex
-	starServer map[string]*notify.StarNotifyS
+	smu        sync.RWMutex
+	starClient map[string]notify.Client
+	starServer map[string]notify.Server
 )
 
 func init() {
-	starClient = make(map[string]*notify.StarNotifyC)
+	starClient = make(map[string]notify.Client)
-	starServer = make(map[string]*notify.StarNotifyS)
+	starServer = make(map[string]notify.Server)
 }
 
-func NewClient(key, netype, value string) (*notify.StarNotifyC, error) {
+func NewClient(key string) notify.Client {
-	client, err := notify.NewNotifyC(netype, value)
+	client := notify.NewClient()
-	if err != nil {
+	cmu.Lock()
-		return client, err
-	}
 	starClient[key] = client
-	return client, err
+	cmu.Unlock()
+	return client
 }
 
-func NewClientWithTimeout(key, netype, value string, timeout time.Duration) (*notify.StarNotifyC, error) {
+func DeleteClient(key string) (err error) {
-	client, err := notify.NewNotifyCWithTimeOut(netype, value, timeout)
+	cmu.RLock()
-	if err != nil {
-		return client, err
-	}
-	starClient[key] = client
-	return client, err
-}
-
-func DeleteClient(key string) error {
 	client, ok := starClient[key]
+	cmu.RUnlock()
 	if !ok {
 		return errors.New("Not Exists Yet!")
 	}
-	if client.Online {
+	if client.Status().Alive {
-		client.ClientStop()
+		err = client.Stop()
 	}
 	client = nil
+	cmu.Lock()
 	delete(starClient, key)
-	return nil
+	cmu.Unlock()
+	return err
 }
 
-func NewServer(key, netype, value string) (*notify.StarNotifyS, error) {
+func NewServer(key string) notify.Server {
-	server, err := notify.NewNotifyS(netype, value)
+	server := notify.NewServer()
-	if err != nil {
+	smu.Lock()
-		return server, err
-	}
 	starServer[key] = server
-	return server, err
+	smu.Unlock()
+	return server
 }
 
 func DeleteServer(key string) error {
+	smu.RLock()
 	server, ok := starServer[key]
+	smu.RUnlock()
 	if !ok {
 		return errors.New("Not Exists Yet!")
 	}
-	if server.Online {
+	if server.Status().Alive {
-		server.ServerStop()
+		server.Stop()
 	}
 	server = nil
+	smu.Lock()
 	delete(starServer, key)
+	smu.Unlock()
 	return nil
 }
 
-func S(key string) *notify.StarNotifyS {
+func S(key string) notify.Server {
+	smu.RLock()
 	server, ok := starServer[key]
+	smu.RUnlock()
 	if !ok {
 		return nil
 	}
 	return server
 }
 
-func C(key string) *notify.StarNotifyC {
+func C(key string) notify.Client {
+	cmu.RLock()
 	client, ok := starClient[key]
+	cmu.RUnlock()
 	if !ok {
 		return nil
 	}
 	return client
 }
 
-func Server(key string) (*notify.StarNotifyS, error) {
+func Server(key string) (notify.Server, error) {
+	smu.RLock()
 	server, ok := starServer[key]
+	smu.RUnlock()
 	if !ok {
 		return nil, errors.New("Not Exists Yet")
 	}
 	return server, nil
 }
 
-func Client(key string) (*notify.StarNotifyC, error) {
+func Client(key string) (notify.Client, error) {
+	cmu.RLock()
 	client, ok := starClient[key]
+	cmu.RUnlock()
 	if !ok {
 		return nil, errors.New("Not Exists Yet")
 	}

v2cs_test.go

@@ -0,0 +1,152 @@
+package notify
+
+import (
+	"fmt"
+	"net"
+	//_ "net/http/pprof"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func Test_ServerTuAndClientCommon(t *testing.T) {
+	//go http.ListenAndServe("0.0.0.0:8888", nil)
+	noEn := func(key, bn []byte) []byte {
+		return bn
+	}
+	_ = noEn
+	server := NewServer()
+	//server.SetDefaultCommDecode(noEn)
+	//server.SetDefaultCommEncode(noEn)
+	err := server.Listen("tcp", "127.0.0.1:12345")
+	if err != nil {
+		panic(err)
+	}
+	server.SetLink("notify", notify)
+	for i := 1; i <= 100; i++ {
+		go func() {
+			client := NewClient()
+			//client.SetMsgEn(noEn)
+			//client.SetMsgDe(noEn)
+			//client.SetSkipExchangeKey(true)
+			err = client.Connect("tcp", "127.0.0.1:12345")
+			if err != nil {
+				t.Fatal(err)
+				time.Sleep(time.Second * 2)
+				return
+			}
+			//client.SetLink("notify", notify)
+			for {
+
+				//nowd = time.Now().UnixNano()
+				client.SendWait("notify", []byte("client hello"), time.Second*15)
+				//client.Send("notify", []byte("client hello"))
+				//time.Sleep(time.Millisecond)
+				//fmt.Println("finished:", float64(time.Now().UnixNano()-nowd)/1000000)
+				//client.Send("notify", []byte("client"))
+			}
+		}()
+	}
+	time.Sleep(time.Second)
+	go func() {
+		time.Sleep(time.Second * 10)
+		server.Stop()
+	}()
+	<-server.StopMonitorChan()
+	fmt.Println(count2)
+}
+
+var count2 int64
+
+func notify(msg *Message) {
+	//fmt.Println(string(msg.Msg.Value))
+	//fmt.Println("called:", float64(time.Now().UnixNano()-nowd)/1000000)
+	if msg.NetType == NET_SERVER {
+		atomic.AddInt64(&count2, 1)
+		msg.Reply([]byte("server reply"))
+	}
+}
+
+func Test_normal(t *testing.T) {
+	server, err := net.Listen("tcp", "127.0.0.1:12345")
+	if err != nil {
+		t.Fatal(err)
+	}
+	go func() {
+		for {
+			conn, err := server.Accept()
+			if err == nil {
+				go func() {
+					for {
+						buf := make([]byte, 1024)
+						_, err := conn.Read(buf)
+						//fmt.Println("S RECV", string(buf[:i]))
+						atomic.AddInt64(&count2, 1)
+						if err == nil {
+							conn.Write([]byte("hello world server"))
+						}
+					}
+				}()
+			}
+		}
+	}()
+	time.Sleep(time.Second * 5)
+	for i := 1; i <= 100; i++ {
+		go func() {
+			conn, err := net.Dial("tcp", "127.0.0.1:12345")
+			if err != nil {
+				panic(err)
+			}
+			for {
+				//nowd = time.Now().UnixNano()
+				_, err := conn.Write([]byte("hello world client"))
+				if err != nil {
+					fmt.Println(err)
+				}
+				buf := make([]byte, 1024)
+				conn.Read(buf)
+
+				continue
+			}
+		}()
+	}
+	time.Sleep(time.Second * 10)
+	fmt.Println(count2)
+}
+
+func Test_normal_udp(t *testing.T) {
+	ludp, _ := net.ResolveUDPAddr("udp", "127.0.0.1:12345")
+	conn, _ := net.ListenUDP("udp", ludp)
+	go func() {
+		for {
+			buf := make([]byte, 1024)
+			_, addr, err := conn.ReadFromUDP(buf)
+			fmt.Println(time.Now(), "S RECV", addr.String())
+			atomic.AddInt64(&count2, 1)
+			if err == nil {
+				conn.WriteToUDP([]byte("hello world server"), addr)
+			}
+		}
+	}()
+	for i := 1; i <= 100; i++ {
+		go func() {
+			conn, err := net.Dial("udp", "127.0.0.1:12345")
+			if err != nil {
+				panic(err)
+			}
+			for {
+				//nowd = time.Now().UnixNano()
+				_, err := conn.Write([]byte("hello world client"))
+				if err != nil {
+					fmt.Println(err)
+				}
+				buf := make([]byte, 1024)
+				conn.Read(buf)
+				fmt.Println(time.Now(), "C RECV")
+				continue
+			}
+		}()
+	}
+	time.Sleep(time.Second * 10)
+	fmt.Println(count2)
+}