update go mod

client.go

@@ -15,7 +15,6 @@ import (
 	"time"
 )
 
-//var nowd int64
 type ClientCommon struct {
 	alive                      atomic.Value
 	status                     Status
@@ -47,6 +46,7 @@ type ClientCommon struct {
 	useHeartBeat               bool
 	sequenceDe                 func([]byte) (interface{}, error)
 	sequenceEn                 func(interface{}) ([]byte, error)
+	debugMode                  bool
 }
 
 func (c *ClientCommon) Connect(network string, addr string) error {
@@ -68,6 +68,16 @@ func (c *ClientCommon) Connect(network string, addr string) error {
 	return c.clientPostInit()
 }
 
+func (c *ClientCommon) DebugMode(dmg bool) {
+	c.mu.Lock()
+	c.debugMode = dmg
+	c.mu.Unlock()
+}
+
+func (c *ClientCommon) IsDebugMode() bool {
+	return c.debugMode
+}
+
 func (c *ClientCommon) ConnectTimeout(network string, addr string, timeout time.Duration) error {
 	if c.alive.Load().(bool) {
 		return errors.New("client already run")
@@ -185,9 +195,14 @@ func (c *ClientCommon) Heartbeat() {
 			c.lastHeartbeat = time.Now().Unix()
 			failedCount = 0
 		}
+		if c.debugMode {
+			fmt.Println("failed to recv heartbeat,timeout!")
+		}
 		failedCount++
 		if failedCount >= 3 {
-			//fmt.Println("heatbeat failed,stop client")
+			if c.debugMode {
+				fmt.Println("heatbeat failed more than 3 times,stop client")
+			}
 			c.alive.Store(false)
 			c.mu.Lock()
 			c.status = Status{
@@ -218,7 +233,9 @@ func (c *ClientCommon) readMessage() {
 		}
 		data := make([]byte, 8192)
 		if c.maxReadTimeout.Seconds() != 0 {
-			c.conn.SetReadDeadline(time.Now().Add(c.maxReadTimeout))
+			if err := c.conn.SetReadDeadline(time.Now().Add(c.maxReadTimeout)); err != nil {
+				//TODO:ALERT
+			}
 		}
 		readNum, err := c.conn.Read(data)
 		if err == os.ErrDeadlineExceeded {
@@ -228,7 +245,7 @@ func (c *ClientCommon) readMessage() {
 			continue
 		}
 		if err != nil {
-			if c.showError {
+			if c.showError || c.debugMode {
 				fmt.Println("client read error", err)
 			}
 			c.alive.Store(false)
@@ -278,7 +295,7 @@ func (c *ClientCommon) loadMessage() {
 				//transfer to Msg
 				msg, err := c.sequenceDe(c.msgDe(c.SecretKey, data.Msg))
 				if err != nil {
-					if c.showError {
+					if c.showError || c.debugMode {
 						fmt.Println("client decode data error", err)
 					}
 					return
@@ -314,8 +331,8 @@ func (c *ClientCommon) dispatchMsg(message Message) {
 			c.noFinSyncMsgPool.Delete(message.ID)
 			return
 		}
-		return
-		//fallthrough
+		//return
+		fallthrough
 	default:
 	}
 	callFn := func(fn func(*Message)) {
@@ -411,7 +428,7 @@ func (c *ClientCommon) sendWait(msg TransferMsg, timeout time.Duration) (Message
 		c.noFinSyncMsgPool.Delete(data.TransferMsg.ID)
 		return Message{}, os.ErrDeadlineExceeded
 	case <-c.stopCtx.Done():
-		return Message{}, errors.New("Service shutdown")
+		return Message{}, errors.New("service shutdown")
 	case msg, ok := <-data.Reply:
 		if !ok {
 			return msg, os.ErrInvalid
@@ -434,7 +451,7 @@ func (c *ClientCommon) sendCtx(msg TransferMsg, ctx context.Context) (Message, e
 		c.noFinSyncMsgPool.Delete(data.TransferMsg.ID)
 		return Message{}, os.ErrDeadlineExceeded
 	case <-c.stopCtx.Done():
-		return Message{}, errors.New("Service shutdown")
+		return Message{}, errors.New("service shutdown")
 	case msg, ok := <-data.Reply:
 		if !ok {
 			return msg, os.ErrInvalid

clienttype.go

@@ -39,6 +39,8 @@ type Client interface {
 	StopMonitorChan() <-chan struct{}
 	Status() Status
 	ShowError(bool)
+	DebugMode(bool)
+	IsDebugMode() bool
 
 	GetSequenceEn() func(interface{}) ([]byte, error)
 	SetSequenceEn(func(interface{}) ([]byte, error))

default.go

@@ -82,5 +82,5 @@ func defaultMsgDe(key []byte, d []byte) []byte {
 }
 
 func init() {
-	Register(TransferMsg{})
+	RegisterName("b612.me/notify.Transfer", TransferMsg{})
 }

go.mod

@@ -3,7 +3,7 @@ module b612.me/notify
 go 1.16
 
 require (
-	b612.me/starcrypto v0.0.1
-	b612.me/stario v0.0.5
-	b612.me/starnet v0.1.3
+	b612.me/starcrypto v0.0.2
+	b612.me/stario v0.0.8
+	b612.me/starnet v0.1.7
 )

go.sum

@@ -1,9 +1,9 @@
-b612.me/starcrypto v0.0.1 h1:xGngzXPUrVbqtWzNw2e+0eWsdG7GG1/X+ONDGIzdriI=
-b612.me/starcrypto v0.0.1/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=
+b612.me/starcrypto v0.0.2 h1:aRf1HcqK8GqHYxLAhWfFC4W/EqQLEFNEmxsBu3wG30o=
+b612.me/starcrypto v0.0.2/go.mod h1:hz0xRnfWNpYOlVrIPoGrQOWPibq4YiUZ7qN5tsQbzPo=
+b612.me/stario v0.0.8 h1:kaA4pszAKLZJm2D9JmiuYSpgjTeE3VaO74vm+H0vBGM=
+b612.me/stario v0.0.8/go.mod h1:or4ssWcxQSjMeu+hRKEgtp0X517b3zdlEOAms8Qscvw=
+b612.me/starnet v0.1.7 h1:k3CUfYNRolC/xw5Ekus2NVWHlqeykSyAH8USGTPKA5o=
+b612.me/starnet v0.1.7/go.mod h1:DNC4i/ezgVLlmxnquf8AeljsL4mQ5vAyxh8vGPQqsys=
 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=

msg.go

@@ -7,7 +7,7 @@ import (
 	"fmt"
 	"net"
 	"os"
-	"sync"
+	"reflect"
 	"sync/atomic"
 	"time"
 )
@@ -45,7 +45,6 @@ type Message struct {
 	ServerConn Client
 	TransferMsg
 	Time time.Time
-	sync.Mutex
 }
 
 type WaitMsg struct {
@@ -475,3 +474,29 @@ func MustToMsgVal(val interface{}) MsgVal {
 	}
 	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

@@ -9,11 +9,22 @@ func Register(data interface{}) {
 	gob.Register(data)
 }
 
+func RegisterName(name string, data interface{}) {
+	gob.RegisterName(name, data)
+}
+
 func RegisterAll(data []interface{}) {
 	for _, v := range data {
 		gob.Register(v)
 	}
 }
+
+func RegisterNames(data map[string]interface{}) {
+	for k, v := range data {
+		gob.RegisterName(k, v)
+	}
+}
+
 func encode(src interface{}) ([]byte, error) {
 	var buf bytes.Buffer
 	enc := gob.NewEncoder(&buf)

server.go

@@ -42,6 +42,7 @@ type ServerCommon struct {
 	sequenceDe                 func([]byte) (interface{}, error)
 	sequenceEn                 func(interface{}) ([]byte, error)
 	showError                  bool
+	debugMode                  bool
 }
 
 func NewServer() Server {
@@ -65,6 +66,19 @@ func NewServer() Server {
 	}
 	return &server
 }
+
+func (s *ServerCommon) DebugMode(dmg bool) {
+	s.mu.Lock()
+	s.debugMode = dmg
+	s.mu.Unlock()
+}
+
+func (s *ServerCommon) IsDebugMode() bool {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.debugMode
+}
+
 func (s *ServerCommon) ShowError(std bool) {
 	s.mu.Lock()
 	s.showError = std
@@ -185,16 +199,22 @@ func (s *ServerCommon) acceptTU() {
 	for {
 		select {
 		case <-s.stopCtx.Done():
+			if s.debugMode {
+				fmt.Println("accept goroutine recv exit signal,exit")
+			}
 			return
 		default:
 		}
 		conn, err := s.listener.Accept()
 		if err != nil {
-			if s.showError {
+			if s.showError || s.debugMode {
 				fmt.Println("error accept:", err)
 			}
 			continue
 		}
+		if s.debugMode {
+			fmt.Println("accept new connection from", conn.RemoteAddr())
+		}
 		var id string
 		for {
 			id = fmt.Sprintf("%s%d%d", conn.RemoteAddr().String(), time.Now().UnixNano(), rand.Int63())
@@ -241,7 +261,7 @@ func (s *ServerCommon) loadMessage() {
 			s.mu.RLock()
 			for _, v := range s.clientPool {
 				wg.Add(1)
-				go func() {
+				go func(v *ClientConn) {
 					defer wg.Done()
 					v.sayGoodByeForTU()
 					v.alive.Store(false)
@@ -252,7 +272,7 @@ func (s *ServerCommon) loadMessage() {
 					}
 					v.stopFn()
 					s.removeClient(v)
-				}()
+				}(v)
 			}
 			s.mu.RUnlock()
 			select {
@@ -282,7 +302,7 @@ func (s *ServerCommon) loadMessage() {
 				//fmt.Println("received:", float64(time.Now().UnixNano()-nowd)/1000000)
 				msg, err := s.sequenceDe(cc.msgDe(cc.SecretKey, data.Msg))
 				if err != nil {
-					if s.showError {
+					if s.showError || s.debugMode {
 						fmt.Println("server decode data error", err)
 					}
 					return
@@ -343,8 +363,8 @@ func (s *ServerCommon) dispatchMsg(message Message) {
 			return
 		}
 		//just throw
-		return
-		//fallthrough
+		//return
+		fallthrough
 	default:
 	}
 	callFn := func(fn func(*Message)) {
@@ -377,7 +397,9 @@ func (s *ServerCommon) sendTU(c *ClientConn, msg TransferMsg) (WaitMsg, error) {
 	data = c.msgEn(c.SecretKey, data)
 	data = s.queue.BuildMessage(data)
 	if c.maxWriteTimeout.Seconds() != 0 {
-		c.tuConn.SetWriteDeadline(time.Now().Add(c.maxWriteTimeout))
+		if err := c.tuConn.SetWriteDeadline(time.Now().Add(c.maxWriteTimeout)); err != nil {
+			return WaitMsg{}, err
+		}
 	}
 	_, err = c.tuConn.Write(data)
 	//fmt.Println("resend:", float64(time.Now().UnixNano()-nowd)/1000000)
@@ -416,7 +438,7 @@ func (s *ServerCommon) sendWait(c *ClientConn, msg TransferMsg, timeout time.Dur
 		s.noFinSyncMsgPool.Delete(data.TransferMsg.ID)
 		return Message{}, os.ErrDeadlineExceeded
 	case <-s.stopCtx.Done():
-		return Message{}, errors.New("Service shutdown")
+		return Message{}, errors.New("service shutdown")
 	case msg, ok := <-data.Reply:
 		if !ok {
 			return msg, os.ErrInvalid
@@ -447,7 +469,7 @@ func (s *ServerCommon) sendCtx(c *ClientConn, msg TransferMsg, ctx context.Conte
 		s.noFinSyncMsgPool.Delete(data.TransferMsg.ID)
 		return Message{}, os.ErrClosed
 	case <-s.stopCtx.Done():
-		return Message{}, errors.New("Service shutdown")
+		return Message{}, errors.New("service shutdown")
 	case msg, ok := <-data.Reply:
 		if !ok {
 			return msg, os.ErrInvalid
@@ -524,6 +546,9 @@ func (s *ServerCommon) acceptUDP() {
 	for {
 		select {
 		case <-s.stopCtx.Done():
+			if s.debugMode {
+				fmt.Println("accept goroutine recv exit signal,exit")
+			}
 			return
 		default:
 		}
@@ -533,6 +558,9 @@ func (s *ServerCommon) acceptUDP() {
 		data := make([]byte, 4096)
 		num, addr, err := s.udpListener.ReadFromUDP(data)
 		id := addr.String()
+		if s.debugMode {
+			fmt.Println("accept new udp message from", id)
+		}
 		//fmt.Println("s recv udp:", float64(time.Now().UnixNano()-nowd)/1000000)
 		s.mu.RLock()
 		if _, ok := s.clientPool[id]; !ok {
@@ -626,7 +654,7 @@ func (s *ServerCommon) GetClient(id string) *ClientConn {
 func (s *ServerCommon) GetClientLists() []*ClientConn {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
-	var list []*ClientConn = make([]*ClientConn, 0, len(s.clientPool))
+	var list = make([]*ClientConn, 0, len(s.clientPool))
 	for _, v := range s.clientPool {
 		list = append(list, v)
 	}

servertype.go

@@ -41,6 +41,8 @@ type Server interface {
 	GetSequenceDe() func([]byte) (interface{}, error)
 	SetSequenceDe(func([]byte) (interface{}, error))
 	ShowError(bool)
+	DebugMode(bool)
+	IsDebugMode() bool
 
 	HeartbeatTimeoutSec() int64
 	SetHeartbeatTimeoutSec(int64)