From c741cd41b8216a63da3d3f0c386627930477e57d Mon Sep 17 00:00:00 2001 From: godotg Date: Thu, 15 Sep 2022 18:19:22 +0800 Subject: [PATCH] =?UTF-8?q?feat[golang]:=20go=E7=9A=84zfoo=E6=9C=8D?= =?UTF-8?q?=E5=8A=A1=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- net/src/test/go/net/codec.go | 20 +-- net/src/test/go/net/codec_test.go | 2 +- net/src/test/go/net/conn.go | 2 +- net/src/test/go/net/message.go | 42 +----- net/src/test/go/net/service_test.go | 2 +- net/src/test/go/znet/codec.go | 65 +++++++++ net/src/test/go/znet/conn.go | 166 +++++++++++++++++++++++ net/src/test/go/znet/def.go | 29 ++++ net/src/test/go/znet/message.go | 38 ++++++ net/src/test/go/znet/service.go | 198 ++++++++++++++++++++++++++++ net/src/test/go/znet/session.go | 36 +++++ net/src/test/go/znet/zne_test.go | 79 +++++++++++ 12 files changed, 616 insertions(+), 63 deletions(-) create mode 100644 net/src/test/go/znet/codec.go create mode 100644 net/src/test/go/znet/conn.go create mode 100644 net/src/test/go/znet/def.go create mode 100644 net/src/test/go/znet/message.go create mode 100644 net/src/test/go/znet/service.go create mode 100644 net/src/test/go/znet/session.go create mode 100644 net/src/test/go/znet/zne_test.go diff --git a/net/src/test/go/net/codec.go b/net/src/test/go/net/codec.go index 17d4dd27..2610c081 100644 --- a/net/src/test/go/net/codec.go +++ b/net/src/test/go/net/codec.go @@ -15,7 +15,6 @@ package net import ( "bytes" "encoding/binary" - "errors" ) // Encode from Message to []byte @@ -34,11 +33,6 @@ func Encode(msg *Message) ([]byte, error) { if err != nil { return nil, err } - err = binary.Write(buffer, binary.LittleEndian, msg.checksum) - if err != nil { - return nil, err - } - return buffer.Bytes(), nil } @@ -62,22 +56,10 @@ func Decode(data []byte) (*Message, error) { return nil, err } - // 检查checksum - var checksum uint32 - err = binary.Read(bufReader, binary.LittleEndian, &checksum) - if err != nil { - return nil, err - } - message := &Message{} message.msgSize = int32(dataSize) message.msgID = msgID message.data = dataBuf - message.checksum = checksum - if message.Verify() { - return message, nil - } - - return nil, errors.New("checksum error") + return message, nil } diff --git a/net/src/test/go/net/codec_test.go b/net/src/test/go/net/codec_test.go index 5c6b0b72..19dadfff 100644 --- a/net/src/test/go/net/codec_test.go +++ b/net/src/test/go/net/codec_test.go @@ -32,5 +32,5 @@ func TestCodec(t *testing.T) { t.Fatal(err) } - t.Logf("ID=%d, Data=%s", msg2.GetID(), string(msg2.GetData())) + t.Logf("ID=%d, Data=%s", msg2.msgID, string(msg2.data)) } diff --git a/net/src/test/go/net/conn.go b/net/src/test/go/net/conn.go index 96dc5342..1f97c155 100644 --- a/net/src/test/go/net/conn.go +++ b/net/src/test/go/net/conn.go @@ -156,7 +156,7 @@ func (c *Conn) readCoroutine(ctx context.Context) { continue } - if msg.GetID() == MsgHeartbeat { + if msg.msgID == MsgHeartbeat { continue } diff --git a/net/src/test/go/net/message.go b/net/src/test/go/net/message.go index a45a531b..c9f658c9 100644 --- a/net/src/test/go/net/message.go +++ b/net/src/test/go/net/message.go @@ -12,10 +12,7 @@ package net import ( - "bytes" - "encoding/binary" "fmt" - "hash/adler32" ) // Message struct @@ -23,7 +20,6 @@ type Message struct { msgSize int32 msgID int32 data []byte - checksum uint32 } // NewMessage create a new message @@ -33,46 +29,10 @@ func NewMessage(msgID int32, data []byte) *Message { msgID: msgID, data: data, } - - msg.checksum = msg.calcChecksum() return msg } -// GetData get message data -func (msg *Message) GetData() []byte { - return msg.data -} - -// GetID get message ID -func (msg *Message) GetID() int32 { - return msg.msgID -} - -// Verify verify checksum -func (msg *Message) Verify() bool { - return msg.checksum == msg.calcChecksum() -} - -func (msg *Message) calcChecksum() uint32 { - if msg == nil { - return 0 - } - - data := new(bytes.Buffer) - - err := binary.Write(data, binary.LittleEndian, msg.msgID) - if err != nil { - return 0 - } - err = binary.Write(data, binary.LittleEndian, msg.data) - if err != nil { - return 0 - } - - checksum := adler32.Checksum(data.Bytes()) - return checksum -} func (msg *Message) String() string { - return fmt.Sprintf("Size=%d ID=%d DataLen=%d Checksum=%d", msg.msgSize, msg.GetID(), len(msg.GetData()), msg.checksum) + return fmt.Sprintf("Size=%d ID=%d DataLen=%d", msg.msgSize, msg.msgID, len(msg.data)) } diff --git a/net/src/test/go/net/service_test.go b/net/src/test/go/net/service_test.go index 81caa21b..d50f3d3b 100644 --- a/net/src/test/go/net/service_test.go +++ b/net/src/test/go/net/service_test.go @@ -47,7 +47,7 @@ func TestService(t *testing.T) { func HandleMessage(s *Session, msg *Message) { fmt.Println("receive msgID:", msg) - fmt.Println("receive data:", string(msg.GetData())) + fmt.Println("receive data:", string(msg.data)) } func HandleDisconnect(s *Session, err error) { diff --git a/net/src/test/go/znet/codec.go b/net/src/test/go/znet/codec.go new file mode 100644 index 00000000..f86cb2be --- /dev/null +++ b/net/src/test/go/znet/codec.go @@ -0,0 +1,65 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package znet + +import ( + "bytes" + "encoding/binary" +) + +// Encode from Message to []byte +func Encode(msg *Message) ([]byte, error) { + buffer := new(bytes.Buffer) + + err := binary.Write(buffer, binary.LittleEndian, msg.msgSize) + if err != nil { + return nil, err + } + err = binary.Write(buffer, binary.LittleEndian, msg.msgID) + if err != nil { + return nil, err + } + err = binary.Write(buffer, binary.LittleEndian, msg.data) + if err != nil { + return nil, err + } + return buffer.Bytes(), nil +} + +// Decode from []byte to Message +func Decode(data []byte) (*Message, error) { + bufReader := bytes.NewReader(data) + + dataSize := len(data) + // 读取消息ID + var msgID int32 + err := binary.Read(bufReader, binary.LittleEndian, &msgID) + if err != nil { + return nil, err + } + + // 读取数据 + dataBufLength := dataSize - 4 - 4 + dataBuf := make([]byte, dataBufLength) + err = binary.Read(bufReader, binary.LittleEndian, &dataBuf) + if err != nil { + return nil, err + } + + message := &Message{} + message.msgSize = int32(dataSize) + message.msgID = msgID + message.data = dataBuf + + return message, nil +} diff --git a/net/src/test/go/znet/conn.go b/net/src/test/go/znet/conn.go new file mode 100644 index 00000000..c1c856e3 --- /dev/null +++ b/net/src/test/go/znet/conn.go @@ -0,0 +1,166 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package znet + +import ( + "bytes" + "context" + "encoding/binary" + "io" + "net" + "time" +) + +// Conn wrap net.Conn +type Conn struct { + sid string + rawConn net.Conn + sendCh chan []byte + done chan error + hbTimer *time.Timer + name string + messageCh chan *Message + hbInterval time.Duration + hbTimeout time.Duration +} + +// GetName Get conn name +func (c *Conn) GetName() string { + return c.name +} + +// NewConn create new conn +func NewConn(c net.Conn, hbInterval time.Duration, hbTimeout time.Duration) *Conn { + conn := &Conn{ + rawConn: c, + sendCh: make(chan []byte, 100), + done: make(chan error), + messageCh: make(chan *Message, 100), + hbInterval: hbInterval, + hbTimeout: hbTimeout, + } + + conn.name = c.RemoteAddr().String() + conn.hbTimer = time.NewTimer(conn.hbInterval) + + if conn.hbInterval == 0 { + conn.hbTimer.Stop() + } + + return conn +} + +// Close close connection +func (c *Conn) Close() { + c.hbTimer.Stop() + c.rawConn.Close() +} + +// SendMessage send message +func (c *Conn) SendMessage(msg *Message) error { + pkg, err := Encode(msg) + if err != nil { + return err + } + + c.sendCh <- pkg + return nil +} + +// writeCoroutine write coroutine +func (c *Conn) writeCoroutine(ctx context.Context) { + hbData := make([]byte, 0) + + for { + select { + case <-ctx.Done(): + return + + case pkt := <-c.sendCh: + + if pkt == nil { + continue + } + + if _, err := c.rawConn.Write(pkt); err != nil { + c.done <- err + } + + case <-c.hbTimer.C: + hbMessage := NewMessage(MsgHeartbeat, hbData) + c.SendMessage(hbMessage) + // 设置心跳timer + if c.hbInterval > 0 { + c.hbTimer.Reset(c.hbInterval) + } + } + } +} + +// readCoroutine read coroutine +func (c *Conn) readCoroutine(ctx context.Context) { + + for { + select { + case <-ctx.Done(): + return + + default: + // 设置超时 + if c.hbInterval > 0 { + err := c.rawConn.SetReadDeadline(time.Now().Add(c.hbTimeout)) + if err != nil { + c.done <- err + continue + } + } + // 读取长度 + buf := make([]byte, 4) + _, err := io.ReadFull(c.rawConn, buf) + if err != nil { + c.done <- err + continue + } + + bufReader := bytes.NewReader(buf) + + var dataSize int32 + err = binary.Read(bufReader, binary.LittleEndian, &dataSize) + if err != nil { + c.done <- err + continue + } + + // 读取数据 + databuf := make([]byte, dataSize) + _, err = io.ReadFull(c.rawConn, databuf) + if err != nil { + c.done <- err + continue + } + + // 解码 + msg, err := Decode(databuf) + if err != nil { + c.done <- err + continue + } + + if msg.msgID == MsgHeartbeat { + continue + } + + c.messageCh <- msg + } + } +} diff --git a/net/src/test/go/znet/def.go b/net/src/test/go/znet/def.go new file mode 100644 index 00000000..65e94808 --- /dev/null +++ b/net/src/test/go/znet/def.go @@ -0,0 +1,29 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package znet + +const ( + // STUnknown Unknown + STUnknown = iota + // STInited Inited + STInited + // STRunning Running + STRunning + // STStop Stop + STStop +) + +const ( + // MsgHeartbeat heartbeat + MsgHeartbeat = iota +) diff --git a/net/src/test/go/znet/message.go b/net/src/test/go/znet/message.go new file mode 100644 index 00000000..ef80c58d --- /dev/null +++ b/net/src/test/go/znet/message.go @@ -0,0 +1,38 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ +package znet + +import ( + "fmt" +) + +// Message struct +type Message struct { + msgSize int32 + msgID int32 + data []byte +} + +// NewMessage create a new message +func NewMessage(msgID int32, data []byte) *Message { + msg := &Message{ + msgSize: int32(len(data)) + 4 + 4, + msgID: msgID, + data: data, + } + return msg +} + + +func (msg *Message) String() string { + return fmt.Sprintf("Size=%d ID=%d DataLen=%d", msg.msgSize, msg.msgID, len(msg.data)) +} diff --git a/net/src/test/go/znet/service.go b/net/src/test/go/znet/service.go new file mode 100644 index 00000000..8b1bf742 --- /dev/null +++ b/net/src/test/go/znet/service.go @@ -0,0 +1,198 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ +package znet + +import ( + "context" + "errors" + "net" + "sync" + "time" +) + +// SocketService struct +type SocketService struct { + onMessage func(*Session, *Message) + onConnect func(*Session) + onDisconnect func(*Session, error) + sessions *sync.Map + hbInterval time.Duration + hbTimeout time.Duration + laddr string + status int + listener net.Listener + stopCh chan error +} + +// NewSocketService create a new socket service +func NewSocketService(laddr string) (*SocketService, error) { + + l, err := net.Listen("tcp", laddr) + + if err != nil { + return nil, err + } + + s := &SocketService{ + sessions: &sync.Map{}, + stopCh: make(chan error), + hbInterval: 0 * time.Second, + hbTimeout: 0 * time.Second, + laddr: laddr, + status: STInited, + listener: l, + } + + return s, nil +} + +// RegMessageHandler register message handler +func (s *SocketService) RegMessageHandler(handler func(*Session, *Message)) { + s.onMessage = handler +} + +// RegConnectHandler register connect handler +func (s *SocketService) RegConnectHandler(handler func(*Session)) { + s.onConnect = handler +} + +// RegDisconnectHandler register disconnect handler +func (s *SocketService) RegDisconnectHandler(handler func(*Session, error)) { + s.onDisconnect = handler +} + +// Serv Start socket service +func (s *SocketService) Serv() { + + s.status = STRunning + ctx, cancel := context.WithCancel(context.Background()) + + defer func() { + s.status = STStop + cancel() + s.listener.Close() + }() + + go s.acceptHandler(ctx) + + for { + select { + + case <-s.stopCh: + return + } + } +} + +func (s *SocketService) acceptHandler(ctx context.Context) { + for { + c, err := s.listener.Accept() + if err != nil { + s.stopCh <- err + return + } + + go s.connectHandler(ctx, c) + } +} + +func (s *SocketService) connectHandler(ctx context.Context, c net.Conn) { + conn := NewConn(c, s.hbInterval, s.hbTimeout) + session := NewSession(conn) + s.sessions.Store(session.sid, session) + + connctx, cancel := context.WithCancel(ctx) + + defer func() { + cancel() + conn.Close() + s.sessions.Delete(session.sid) + }() + + go conn.readCoroutine(connctx) + go conn.writeCoroutine(connctx) + + if s.onConnect != nil { + s.onConnect(session) + } + + for { + select { + case err := <-conn.done: + + if s.onDisconnect != nil { + s.onDisconnect(session, err) + } + return + + case msg := <-conn.messageCh: + if s.onMessage != nil { + s.onMessage(session, msg) + } + } + } +} + +// GetStatus get socket service status +func (s *SocketService) GetStatus() int { + return s.status +} + +// Stop stop socket service with reason +func (s *SocketService) Stop(reason string) { + s.stopCh <- errors.New(reason) +} + +// SetHeartBeat set heart beat +func (s *SocketService) SetHeartBeat(hbInterval time.Duration, hbTimeout time.Duration) error { + if s.status == STRunning { + return errors.New("Can't set heart beat on service running") + } + + s.hbInterval = hbInterval + s.hbTimeout = hbTimeout + + return nil +} + +// GetConnsCount get connect count +func (s *SocketService) GetConnsCount() int { + var count int + s.sessions.Range(func(k, v interface{}) bool { + count++ + return true + }) + return count +} + +// Unicast Unicast with session ID +func (s *SocketService) Unicast(sid string, msg *Message) { + v, ok := s.sessions.Load(sid) + if ok { + session := v.(*Session) + err := session.conn.SendMessage(msg) + if err != nil { + return + } + } +} + +// Broadcast Broadcast to all connections +func (s *SocketService) Broadcast(msg *Message) { + s.sessions.Range(func(k, v interface{}) bool { + s := v.(*Session) + if err := s.conn.SendMessage(msg); err != nil { + // log.Println(err) + } + return true + }) +} diff --git a/net/src/test/go/znet/session.go b/net/src/test/go/znet/session.go new file mode 100644 index 00000000..b145c00f --- /dev/null +++ b/net/src/test/go/znet/session.go @@ -0,0 +1,36 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ +package znet + +import "sync/atomic" + +// Session struct +type Session struct { + sid uint64 + uid uint64 + conn *Conn +} + +var uuid uint64 + +// NewSession create a new session +func NewSession(conn *Conn) *Session { + var suuid = atomic.AddUint64(&uuid, 1) + + session := &Session{ + sid: suuid, + uid: 0,// 可以为用户的id + conn: conn, + } + + return session +} diff --git a/net/src/test/go/znet/zne_test.go b/net/src/test/go/znet/zne_test.go new file mode 100644 index 00000000..f0d37afe --- /dev/null +++ b/net/src/test/go/znet/zne_test.go @@ -0,0 +1,79 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ +package znet + +import ( + "fmt" + "net" + "testing" + "time" +) + +func TestService(t *testing.T) { + host := "127.0.0.1:18787" + + ss, err := NewSocketService(host) + if err != nil { + return + } + + // ss.SetHeartBeat(5*time.Second, 30*time.Second) + + ss.RegMessageHandler(HandleMessage) + ss.RegConnectHandler(HandleConnect) + ss.RegDisconnectHandler(HandleDisconnect) + + go NewClientConnect() + + timer := time.NewTimer(time.Second * 1) + go func() { + <-timer.C + ss.Stop("stop service") + t.Log("service stoped") + }() + + t.Log("service running on " + host) + ss.Serv() +} + +func HandleMessage(s *Session, msg *Message) { + fmt.Println("receive msgID:", msg) + fmt.Println("receive data:", string(msg.data)) +} + +func HandleDisconnect(s *Session, err error) { + fmt.Println(s.conn.GetName() + " lost.") +} + +func HandleConnect(s *Session) { + fmt.Println(s.conn.GetName() + " connected.") +} + +func NewClientConnect() { + host := "127.0.0.1:18787" + tcpAddr, err := net.ResolveTCPAddr("tcp", host) + if err != nil { + return + } + + conn, err := net.DialTCP("tcp", nil, tcpAddr) + if err != nil { + return + } + + msg := NewMessage(1, []byte("Hello Zero!")) + data, err := Encode(msg) + if err != nil { + return + } + conn.Write(data) +}