optimize code & adjuste directory (#105)

Allenxuxu · Nov 9, 2021 · 6d40211 · 6d40211
1 parent de840aa
commit 6d40211
Show file tree

Hide file tree

Showing 32 changed files with 171 additions and 193 deletions.
diff --git a/benchmarks/gev-echo-server/echo.go b/benchmarks/gev-echo-server/echo.go
@@ -9,14 +9,13 @@ import (
 	"github.com/Allenxuxu/gev/log"
 
 	"github.com/Allenxuxu/gev"
-	"github.com/Allenxuxu/gev/connection"
 )
 
 type example struct {
 }
 
-func (s *example) OnConnect(c *connection.Connection) {}
-func (s *example) OnMessage(c *connection.Connection, ctx interface{}, data []byte) (out interface{}) {
+func (s *example) OnConnect(c *gev.Connection) {}
+func (s *example) OnMessage(c *gev.Connection, ctx interface{}, data []byte) (out interface{}) {
 
 	out = data
 
@@ -29,7 +28,7 @@ func (s *example) OnMessage(c *connection.Connection, ctx interface{}, data []by
 	return
 }
 
-func (s *example) OnClose(c *connection.Connection) {
+func (s *example) OnClose(c *gev.Connection) {
 	//log.Error("onclose ")
 }
 

diff --git a/benchmarks/websocket/server.go b/benchmarks/websocket/server.go
@@ -7,7 +7,6 @@ import (
 	"github.com/Allenxuxu/gev/log"
 
 	"github.com/Allenxuxu/gev"
-	"github.com/Allenxuxu/gev/connection"
 	"github.com/Allenxuxu/gev/plugins/websocket"
 	"github.com/Allenxuxu/gev/plugins/websocket/ws"
 )
@@ -18,11 +17,11 @@ type example struct {
 // connection lifecycle
 // OnConnect() -> OnRequest() -> OnHeader() -> OnMessage() -> OnClose()
 
-func (s *example) OnConnect(c *connection.Connection) {
+func (s *example) OnConnect(c *gev.Connection) {
 	//log.Println("OnConnect: ", c.PeerAddr())
 }
 
-func (s *example) OnMessage(c *connection.Connection, data []byte) (messageType ws.MessageType, out []byte) {
+func (s *example) OnMessage(c *gev.Connection, data []byte) (messageType ws.MessageType, out []byte) {
 	//log.Println("OnMessage: ", string(data))
 
 	messageType = ws.MessageBinary
@@ -31,13 +30,13 @@ func (s *example) OnMessage(c *connection.Connection, data []byte) (messageType
 	return
 }
 
-func (s *example) OnClose(c *connection.Connection) {
+func (s *example) OnClose(c *gev.Connection) {
 	//log.Println("123 OnClose", c.PeerAddr())
 }
 
 // NewWebSocketServer 创建 WebSocket Server
 func NewWebSocketServer(handler websocket.WSHandler, u *ws.Upgrader, opts ...gev.Option) (server *gev.Server, err error) {
-	opts = append(opts, gev.Protocol(websocket.New(u)))
+	opts = append(opts, gev.CustomProtocol(websocket.New(u)))
 	return gev.NewServer(websocket.NewHandlerWrap(u, handler), opts...)
 }
 

diff --git a/connection/connection.go → connection.go b/connection/connection.go → connection.go
@@ -1,4 +1,4 @@
-package connection
+package gev
 
 import (
 	"errors"
@@ -45,8 +45,8 @@ type Connection struct {
 
 var ErrConnectionClosed = errors.New("connection closed")
 
-// New 创建 Connection
-func New(fd int,
+// NewConnection 创建 Connection
+func NewConnection(fd int,
 	loop *eventloop.EventLoop,
 	sa unix.Sockaddr,
 	protocol Protocol,
@@ -112,12 +112,12 @@ func (c *Connection) Connected() bool {
 }
 
 // Send 用来在非 loop 协程发送
-func (c *Connection) Send(data interface{}, opts ...Option) error {
+func (c *Connection) Send(data interface{}, opts ...ConnectionOption) error {
 	if !c.connected.Get() {
 		return ErrConnectionClosed
 	}
 
-	opt := Options{}
+	opt := ConnectionOptions{}
 	for _, o := range opts {
 		o(&opt)
 	}
@@ -271,7 +271,7 @@ func (c *Connection) handleWrite(fd int) (closed bool) {
 
 	if c.outBuffer.IsEmpty() {
 		if err := c.loop.EnableRead(fd); err != nil {
-			log.Error("[EnableRead]", err)
+			log.Error("[enableRead]", err)
 		}
 	}
 

diff --git a/connection/options.go b/connection/options.go
diff --git a/connection_options.go b/connection_options.go
@@ -0,0 +1,15 @@
+package gev
+
+type SendInLoopFunc func(interface{})
+
+type ConnectionOptions struct {
+	sendInLoopFinish SendInLoopFunc
+}
+
+type ConnectionOption func(*ConnectionOptions)
+
+func SendInLoop(f SendInLoopFunc) ConnectionOption {
+	return func(o *ConnectionOptions) {
+		o.sendInLoopFinish = f
+	}
+}
diff --git a/connection/context.go → context.go b/connection/context.go → context.go
@@ -1,4 +1,4 @@
-package connection
+package gev
 
 import "sync"
 

diff --git a/connection/context_test.go → context_test.go b/connection/context_test.go → context_test.go
@@ -1,4 +1,4 @@
-package connection
+package gev
 
 import (
 	"fmt"

diff --git a/eventloop/eventloop.go b/eventloop/eventloop.go
@@ -83,7 +83,7 @@ func (l *EventLoop) DeleteFdInLoop(fd int) {
 	l.ConnCunt.Add(-1)
 }
 
-// AddSocketAndEnableRead 增加 Socket 到时间循环中，并注册可读事件
+// AddSocketAndEnableRead 增加 Socket 到事件循环中，并注册可读事件
 func (l *EventLoop) AddSocketAndEnableRead(fd int, s Socket) error {
 	l.sockets[fd] = s
 	if err := l.poll.AddRead(fd); err != nil {
@@ -105,8 +105,8 @@ func (l *EventLoop) EnableRead(fd int) error {
 	return l.poll.EnableRead(fd)
 }
 
-// RunLoop 启动事件循环
-func (l *EventLoop) RunLoop() {
+// Run 启动事件循环
+func (l *EventLoop) Run() {
 	l.poll.Poll(l.handlerEvent)
 }
 

diff --git a/eventloop/eventloop_test.go b/eventloop/eventloop_test.go
@@ -30,7 +30,7 @@ func TestEventLoop_RunLoop(t *testing.T) {
 		}
 	}()
 
-	el.RunLoop()
+	el.Run()
 }
 
 func TestEventLoopSize(t *testing.T) {

diff --git a/example/bufferlength/main.go b/example/bufferlength/main.go
@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"github.com/Allenxuxu/gev"
-	"github.com/Allenxuxu/gev/connection"
 )
 
 const clientsKey = "demo_push_message_key"
@@ -54,7 +53,7 @@ func (s *Server) RunPush() {
 	for e := s.conn.Front(); e != nil; e = next {
 		next = e.Next()
 
-		c := e.Value.(*connection.Connection)
+		c := e.Value.(*gev.Connection)
 		if c.WriteBufferLength() > 1024*10 {
 			log.Printf("write buffer length > 1024*10")
 			continue
@@ -64,7 +63,7 @@ func (s *Server) RunPush() {
 }
 
 // OnConnect callback
-func (s *Server) OnConnect(c *connection.Connection) {
+func (s *Server) OnConnect(c *gev.Connection) {
 	log.Println(" OnConnect ： ", c.PeerAddr())
 
 	s.mu.Lock()
@@ -74,15 +73,15 @@ func (s *Server) OnConnect(c *connection.Connection) {
 }
 
 // OnMessage callback
-func (s *Server) OnMessage(c *connection.Connection, ctx interface{}, data []byte) (out interface{}) {
+func (s *Server) OnMessage(c *gev.Connection, ctx interface{}, data []byte) (out interface{}) {
 	log.Printf("OnMessage, read buffer len %d, write buffer len %d \n", c.ReadBufferLength(), c.WriteBufferLength())
 
 	out = data
 	return
 }
 
 // OnClose callback
-func (s *Server) OnClose(c *connection.Connection) {
+func (s *Server) OnClose(c *gev.Connection) {
 	log.Println("OnClose")
 	v, ok := c.Get(clientsKey)
 	if !ok {

diff --git a/example/echo/echo.go b/example/echo/echo.go
@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/Allenxuxu/gev"
-	"github.com/Allenxuxu/gev/connection"
 	"github.com/Allenxuxu/gev/log"
 	"github.com/Allenxuxu/toolkit/sync/atomic"
 )
@@ -17,17 +16,17 @@ type example struct {
 	Count atomic.Int64
 }
 
-func (s *example) OnConnect(c *connection.Connection) {
+func (s *example) OnConnect(c *gev.Connection) {
 	s.Count.Add(1)
 	//log.Println(" OnConnect ： ", c.PeerAddr())
 }
-func (s *example) OnMessage(c *connection.Connection, ctx interface{}, data []byte) (out interface{}) {
+func (s *example) OnMessage(c *gev.Connection, ctx interface{}, data []byte) (out interface{}) {
 	//log.Println("OnMessage")
 	out = data
 	return
 }
 
-func (s *example) OnClose(c *connection.Connection) {
+func (s *example) OnClose(c *gev.Connection) {
 	s.Count.Add(-1)
 	//log.Println("OnClose")
 }

diff --git a/example/idleconnection/echo.go b/example/idleconnection/echo.go
@@ -8,23 +8,22 @@ import (
 	"time"
 
 	"github.com/Allenxuxu/gev"
-	"github.com/Allenxuxu/gev/connection"
 	"github.com/Allenxuxu/gev/log"
 )
 
 type example struct {
 }
 
-func (s *example) OnConnect(c *connection.Connection) {
+func (s *example) OnConnect(c *gev.Connection) {
 	log.Info(" OnConnect ： ", c.PeerAddr())
 }
-func (s *example) OnMessage(c *connection.Connection, ctx interface{}, data []byte) (out interface{}) {
+func (s *example) OnMessage(c *gev.Connection, ctx interface{}, data []byte) (out interface{}) {
 	log.Infof("OnMessage from : %s", c.PeerAddr())
 	out = data
 	return
 }
 
-func (s *example) OnClose(c *connection.Connection) {
+func (s *example) OnClose(c *gev.Connection) {
 	log.Info("OnClose: ", c.PeerAddr())
 }
 

diff --git a/example/maxconnection/main.go b/example/maxconnection/main.go
@@ -6,7 +6,6 @@ import (
 	_ "net/http/pprof"
 
 	"github.com/Allenxuxu/gev"
-	"github.com/Allenxuxu/gev/connection"
 	"github.com/Allenxuxu/toolkit/sync/atomic"
 )
 
@@ -42,7 +41,7 @@ func (s *Server) Stop() {
 }
 
 // OnConnect callback
-func (s *Server) OnConnect(c *connection.Connection) {
+func (s *Server) OnConnect(c *gev.Connection) {
 	s.clientNum.Add(1)
 	log.Println(" OnConnect ： ", c.PeerAddr())
 
@@ -54,14 +53,14 @@ func (s *Server) OnConnect(c *connection.Connection) {
 }
 
 // OnMessage callback
-func (s *Server) OnMessage(c *connection.Connection, ctx interface{}, data []byte) (out interface{}) {
+func (s *Server) OnMessage(c *gev.Connection, ctx interface{}, data []byte) (out interface{}) {
 	log.Println("OnMessage")
 	out = data
 	return
 }
 
 // OnClose callback
-func (s *Server) OnClose(c *connection.Connection) {
+func (s *Server) OnClose(c *gev.Connection) {
 	s.clientNum.Add(-1)
 	log.Println("OnClose")
 }

diff --git a/example/protobuf/server/main.go b/example/protobuf/server/main.go
@@ -6,18 +6,17 @@ import (
 	"strconv"
 
 	"github.com/Allenxuxu/gev"
-	"github.com/Allenxuxu/gev/connection"
 	pb "github.com/Allenxuxu/gev/example/protobuf/proto"
 	"github.com/Allenxuxu/gev/plugins/protobuf"
 	"google.golang.org/protobuf/proto"
 )
 
 type example struct{}
 
-func (s *example) OnConnect(c *connection.Connection) {
+func (s *example) OnConnect(c *gev.Connection) {
 	log.Println(" OnConnect ： ", c.PeerAddr())
 }
-func (s *example) OnMessage(c *connection.Connection, ctx interface{}, data []byte) (out interface{}) {
+func (s *example) OnMessage(c *gev.Connection, ctx interface{}, data []byte) (out interface{}) {
 	msgType := ctx.(string)
 
 	switch msgType {
@@ -40,7 +39,7 @@ func (s *example) OnMessage(c *connection.Connection, ctx interface{}, data []by
 	return
 }
 
-func (s *example) OnClose(c *connection.Connection) {
+func (s *example) OnClose(c *gev.Connection) {
 	log.Println("OnClose")
 }
 
@@ -57,7 +56,7 @@ func main() {
 		gev.Network("tcp"),
 		gev.Address(":"+strconv.Itoa(port)),
 		gev.NumLoops(loops),
-		gev.Protocol(&protobuf.Protocol{}))
+		gev.CustomProtocol(&protobuf.Protocol{}))
 	if err != nil {
 		panic(err)
 	}

diff --git a/example/protocol/protocol.go b/example/protocol/protocol.go
@@ -3,7 +3,7 @@ package main
 import (
 	"encoding/binary"
 
-	"github.com/Allenxuxu/gev/connection"
+	"github.com/Allenxuxu/gev"
 	"github.com/Allenxuxu/ringbuffer"
 	"github.com/gobwas/pool/pbytes"
 )
@@ -12,7 +12,7 @@ const exampleHeaderLen = 4
 
 type ExampleProtocol struct{}
 
-func (d *ExampleProtocol) UnPacket(c *connection.Connection, buffer *ringbuffer.RingBuffer) (interface{}, []byte) {
+func (d *ExampleProtocol) UnPacket(c *gev.Connection, buffer *ringbuffer.RingBuffer) (interface{}, []byte) {
 	if buffer.VirtualLength() > exampleHeaderLen {
 		buf := pbytes.GetLen(exampleHeaderLen)
 		defer pbytes.Put(buf)
@@ -32,7 +32,7 @@ func (d *ExampleProtocol) UnPacket(c *connection.Connection, buffer *ringbuffer.
 	return nil, nil
 }
 
-func (d *ExampleProtocol) Packet(c *connection.Connection, data interface{}) []byte {
+func (d *ExampleProtocol) Packet(c *gev.Connection, data interface{}) []byte {
 	dd := data.([]byte)
 	dataLen := len(dd)
 	ret := make([]byte, exampleHeaderLen+dataLen)