add static func for encode message to byte (#39)

Author: szpnygo

agent/agent.go

@@ -21,6 +21,7 @@ type Agent struct {
 	conn             acceptor.AcceptorConn       // the Conn
 	agentLog         *zap.Logger                 // logger
 	chSend           chan *bytebuffer.ByteBuffer // send message channel
+	chSendByte       chan []byte                 // send message byte channel
 	chStopWrite      chan struct{}               // close message send
 	chStopHeartbeat  chan struct{}               // close heartbeat
 	chDie            chan struct{}               // wait for close
@@ -108,6 +109,10 @@ func (a *Agent) write() {
 				return
 			}
 			bytebuffer.Put(msg)
+		case msg := <-a.chSendByte:
+			if _, err := a.conn.Write(msg); err != nil {
+				return
+			}
 		case <-a.chStopWrite:
 			return
 		}
@@ -179,10 +184,8 @@ func (a *Agent) Send(in interface{}, name ...string) error {
 }
 
 func (a *Agent) SendData(data []byte) {
-	b := bytebuffer.Get()
-	b.Set(data)
 	select {
-	case a.chSend <- b:
+	case a.chSendByte <- data:
 	case <-a.chDie:
 	}
 }

agent/factory.go

@@ -60,6 +60,7 @@ func (af *AgentFacotry) NewAgent(conn acceptor.AcceptorConn) *Agent {
 		conn:             conn,
 		agentLog:         gslog.NewLog("agent").With(zap.Int64("agent_id", agentId)),
 		chSend:           make(chan *bytebuffer.ByteBuffer, af.messagesBufferSize),
+		chSendByte:       make(chan []byte, af.messagesBufferSize),
 		chStopWrite:      make(chan struct{}),
 		chStopHeartbeat:  make(chan struct{}),
 		chDie:            make(chan struct{}),

app.go

@@ -71,6 +71,7 @@ func NewApp(config *config.Config) *App {
 		MessageServer: appBuidler.messageServer,
 		webServer:     appBuidler.webServer,
 	}
+	message.DefaultMessageServer = app.MessageServer
 
 	//初始化便捷服务
 	helper := newAppHelper(app)

message/static.go

@@ -0,0 +1,11 @@
+package message
+
+var DefaultMessageServer *MessageServer
+
+func EncodeMessage(in interface{}, name ...string) ([]byte, error) {
+	packet, err := DefaultMessageServer.EncodeMessage(in, name...)
+	if err != nil {
+		return nil, err
+	}
+	return packet.ToByte(), nil
+}

packet/packet.go

@@ -81,14 +81,32 @@ func (p *Packet) ToData() *bytebuffer.ByteBuffer {
 	}()
 
 	b := bytebuffer.Get()
-	if p.HeaderByte[0] == HeaderFlag {
+	if len(p.HeaderByte) > 0 && p.HeaderByte[0] == HeaderFlag {
 		_, _ = b.Write(p.HeaderByte)
 	}
 	_, _ = b.Write(p.Data)
 
 	return b
 }
 
+func (p *Packet) ToByte() []byte {
+	defer func() {
+		PutPacket(p)
+	}()
+
+	length := len(p.Data)
+	if len(p.HeaderByte) > 0 && p.HeaderByte[0] == HeaderFlag {
+		length += HeaderLength
+		data := make([]byte, length)
+		copy(data, p.HeaderByte)
+		copy(data[HeaderLength:], p.Data)
+		return data
+	}
+	data := make([]byte, length)
+	copy(data, p.Data)
+	return data
+}
+
 // packetType: 2bit  // 0x01: system, 0x02: service
 // module:     6bit
 // action:     16bit

packet/packet_test.go

@@ -318,6 +318,26 @@ func TestPacket_ToData(t *testing.T) {
 				B: []byte{0x7E, 0x0A, 0x41, 0x00, 0x02, 0x00, 0x00, 0x0F, 0x0A, 0x0D, 0x31, 0x36, 0x36, 0x30, 0x33, 0x36, 0x30, 0x39, 0x31, 0x35, 0x35, 0x35, 0x36},
 			},
 		},
+		{
+			name: "to data2",
+			fields: fields{
+				HeaderByte: []byte{},
+				Data:       []byte{0x0A, 0x0D, 0x31, 0x36, 0x36, 0x30, 0x33, 0x36, 0x30, 0x39, 0x31, 0x35, 0x35, 0x35, 0x36},
+			},
+			want: &bytebuffer.ByteBuffer{
+				B: []byte{0x0A, 0x0D, 0x31, 0x36, 0x36, 0x30, 0x33, 0x36, 0x30, 0x39, 0x31, 0x35, 0x35, 0x35, 0x36},
+			},
+		},
+		{
+			name: "to data2",
+			fields: fields{
+				HeaderByte: []byte{0x20},
+				Data:       []byte{0x0A, 0x0D, 0x31, 0x36, 0x36, 0x30, 0x33, 0x36, 0x30, 0x39, 0x31, 0x35, 0x35, 0x35, 0x36},
+			},
+			want: &bytebuffer.ByteBuffer{
+				B: []byte{0x0A, 0x0D, 0x31, 0x36, 0x36, 0x30, 0x33, 0x36, 0x30, 0x39, 0x31, 0x35, 0x35, 0x35, 0x36},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -329,6 +349,9 @@ func TestPacket_ToData(t *testing.T) {
 			if got := p.ToData(); !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("Packet.ToData() = %v, want %v", got, tt.want)
 			}
+			if !reflect.DeepEqual(p.ToData().B, p.ToByte()) {
+				t.Errorf("Packet.ToData() = %v, want %v", p.ToByte(), p.ToData().B)
+			}
 		})
 	}
 }

session/static.go

@@ -1,6 +1,8 @@
 package session
 
-import "github.com/metagogs/gogs/utils/slicex"
+import (
+	"github.com/metagogs/gogs/utils/slicex"
+)
 
 var DefaultSessionPool SessionPool
 
@@ -21,6 +23,12 @@ func SendMessageByID(sessionId int64, in interface{}) {
 	}
 }
 
+func SendDataByID(sessionId int64, in []byte) {
+	if sess, err := DefaultSessionPool.GetSessionByID(sessionId); err == nil {
+		sess.SendData(in)
+	}
+}
+
 func ListSessions() []*Session {
 	return DefaultSessionPool.ListSessions()
 }
@@ -37,3 +45,14 @@ func BroadcastMessage(users []string, send interface{}, filter *SessionFilter, e
 		}
 	}
 }
+
+func BroadcastData(users []string, send []byte, filter *SessionFilter, exclude ...string) {
+	for _, u := range users {
+		if slicex.InSlice(u, exclude) {
+			continue
+		}
+		if result, _ := GetSessionByUID(u, filter); len(result) > 0 {
+			SendDataByID(result[0], send)
+		}
+	}
+}