feat[golang]: go的zfoo服务器

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
 }

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

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
 			}
 

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

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

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

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

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

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

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

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

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