TestService: wait for OnConnect and OnDisconnect

.github/workflows/tests.yaml

@@ -23,4 +23,4 @@ jobs:
         with:
           go-version: '1.21.x'
       - name: Test with the Go CLI
-        run: go test -timeout 10s -v ./...
+        run: go test -v ./...

cnet/service_test.go

@@ -1,97 +0,0 @@
-package cnet_test
-
-import (
-	"context"
-	"fmt"
-	"log"
-	"net"
-	"os"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/CPunch/gopenfusion/cnet"
-	"github.com/CPunch/gopenfusion/cnet/protocol"
-	"github.com/CPunch/gopenfusion/util"
-	"github.com/matryer/is"
-)
-
-var (
-	srvcPort int
-)
-
-const (
-	timeout       = 2 * time.Second
-	maxDummyPeers = 5
-)
-
-func TestMain(m *testing.M) {
-	var err error
-	srvcPort, err = cnet.RandomPort()
-	if err != nil {
-		panic(err)
-	}
-
-	os.Exit(m.Run())
-}
-
-func TestService(t *testing.T) {
-	is := is.New(t)
-	ctx, cancel := context.WithCancel(context.Background())
-	srvc := cnet.NewService(ctx, "TEST", srvcPort)
-	wg := sync.WaitGroup{}
-
-	// shutdown service when test is done
-	defer func() {
-		cancel()
-		is.True(util.SelectWithTimeout(srvc.Stopped(), timeout)) // wait for service to stop with timeout
-	}()
-
-	// our dummy packet handler
-	wg.Add(maxDummyPeers)
-	srvc.AddPacketHandler(0x1234, func(peer *cnet.Peer, pkt protocol.Packet) error {
-		log.Printf("Received packet %#v", pkt)
-		wg.Done()
-		return nil
-	})
-
-	// wait for all dummy peers to connect and disconnect
-	wg.Add(maxDummyPeers)
-	srvc.OnConnect = func(peer *cnet.Peer) {
-		wg.Done()
-	}
-
-	wg.Add(maxDummyPeers)
-	srvc.OnDisconnect = func(peer *cnet.Peer) {
-		wg.Done()
-	}
-
-	// run service
-	go func() { is.NoErr(srvc.Start()) }()                   // srvc.Start error
-	is.True(util.SelectWithTimeout(srvc.Started(), timeout)) // wait for service to start with timeout
-
-	wg.Add(maxDummyPeers * 2) // 2 wg.Done() per peer for receiving packets
-	for i := 0; i < maxDummyPeers; i++ {
-		go func() {
-			// make dummy client
-			conn, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", srvcPort))
-			is.NoErr(err) // net.Dial error
-
-			peer := cnet.NewPeer(ctx, conn)
-			go func() {
-				defer peer.Kill()
-
-				// send dummy packets
-				for i := 0; i < 2; i++ {
-					is.NoErr(peer.Send(0x1234)) // peer.Send error
-				}
-			}()
-
-			// we wait until Handler gracefully exits (peer was killed)
-			peer.Handler(make(chan *cnet.PacketEvent))
-			wg.Done()
-		}()
-	}
-
-	is.True(util.WaitWithTimeout(&wg, timeout)) // wait for all dummy peers to be done with timeout
-}

internal/db/account.go

@@ -6,7 +6,7 @@ import (
 
 	"golang.org/x/crypto/bcrypt"
 
-	"github.com/CPunch/gopenfusion/cnet/protocol"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 	"github.com/georgysavva/scany/v2/sqlscan"
 )
 

internal/db/db_test.go

@@ -8,8 +8,8 @@ import (
 
 	"github.com/matryer/is"
 
-	"github.com/CPunch/gopenfusion/cnet/protocol"
 	"github.com/CPunch/gopenfusion/internal/db"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 	"github.com/bitcomplete/sqltestutil"
 )
 

internal/db/inventory.go

@@ -3,7 +3,7 @@ package db
 import (
 	"database/sql"
 
-	"github.com/CPunch/gopenfusion/cnet/protocol"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 )
 
 type Inventory struct {

internal/db/players.go

@@ -3,8 +3,8 @@ package db
 import (
 	"database/sql"
 
-	"github.com/CPunch/gopenfusion/cnet/protocol"
 	"github.com/CPunch/gopenfusion/config"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 	"github.com/blockloop/scan"
 )
 

internal/entity/chunk.go

@@ -3,22 +3,8 @@ package entity
 import (
 	"log"
 	"sync"
-
-	"github.com/CPunch/gopenfusion/config"
 )
 
-type ChunkPosition struct {
-	X int
-	Y int
-}
-
-func MakeChunkPosition(x, y int) ChunkPosition {
-	return ChunkPosition{
-		X: x / (config.VIEW_DISTANCE / 3),
-		Y: y / (config.VIEW_DISTANCE / 3),
-	}
-}
-
 type Chunk struct {
 	Position ChunkPosition
 	entities map[Entity]struct{}

internal/entity/chunkposition.go

@@ -0,0 +1,15 @@
+package entity
+
+import "github.com/CPunch/gopenfusion/config"
+
+type ChunkPosition struct {
+	X int
+	Y int
+}
+
+func MakeChunkPosition(x, y int) ChunkPosition {
+	return ChunkPosition{
+		X: x / (config.VIEW_DISTANCE / 3),
+		Y: y / (config.VIEW_DISTANCE / 3),
+	}
+}

internal/entity/entity.go

@@ -1,6 +1,6 @@
 package entity
 
-import "github.com/CPunch/gopenfusion/cnet"
+import "github.com/CPunch/gopenfusion/internal/protocol"
 
 type EntityKind int
 
@@ -20,6 +20,6 @@ type Entity interface {
 	SetPosition(x, y, z int)
 	SetAngle(angle int)
 
-	DisappearFromViewOf(peer *cnet.Peer)
-	EnterIntoViewOf(peer *cnet.Peer)
+	DisappearFromViewOf(peer *protocol.CNPeer)
+	EnterIntoViewOf(peer *protocol.CNPeer)
 }

internal/entity/entity_test.go

@@ -3,7 +3,7 @@ package entity_test
 import (
 	"testing"
 
-	"github.com/CPunch/gopenfusion/shard/entity"
+	"github.com/CPunch/gopenfusion/internal/entity"
 	"github.com/matryer/is"
 )
 

internal/entity/npc.go

@@ -3,8 +3,7 @@ package entity
 import (
 	"sync/atomic"
 
-	"github.com/CPunch/gopenfusion/cnet"
-	"github.com/CPunch/gopenfusion/cnet/protocol"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 )
 
 type NPC struct {
@@ -63,13 +62,13 @@ func (npc *NPC) SetAngle(angle int) {
 	npc.Angle = angle
 }
 
-func (npc *NPC) DisappearFromViewOf(peer *cnet.Peer) {
+func (npc *NPC) DisappearFromViewOf(peer *protocol.CNPeer) {
 	peer.Send(protocol.P_FE2CL_NPC_EXIT, protocol.SP_FE2CL_NPC_EXIT{
 		INPC_ID: int32(npc.ID),
 	})
 }
 
-func (npc *NPC) EnterIntoViewOf(peer *cnet.Peer) {
+func (npc *NPC) EnterIntoViewOf(peer *protocol.CNPeer) {
 	peer.Send(protocol.P_FE2CL_NPC_NEW, protocol.SP_FE2CL_NPC_NEW{
 		NPCAppearanceData: npc.GetAppearanceData(),
 	})

internal/entity/player.go

@@ -1,18 +1,17 @@
 package entity
 
 import (
-	"github.com/CPunch/gopenfusion/cnet"
-	"github.com/CPunch/gopenfusion/cnet/protocol"
 	"github.com/CPunch/gopenfusion/internal/db"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 )
 
 type Player struct {
 	db.Player
-	Peer  *cnet.Peer
+	Peer  *protocol.CNPeer
 	Chunk ChunkPosition
 }
 
-func NewPlayer(peer *cnet.Peer, player *db.Player) *Player {
+func NewPlayer(peer *protocol.CNPeer, player *db.Player) *Player {
 	return &Player{
 		Player: *player,
 		Peer:   peer,
@@ -52,13 +51,13 @@ func (plr *Player) SetAngle(angle int) {
 	plr.Angle = angle
 }
 
-func (plr *Player) DisappearFromViewOf(peer *cnet.Peer) {
+func (plr *Player) DisappearFromViewOf(peer *protocol.CNPeer) {
 	peer.Send(protocol.P_FE2CL_PC_EXIT, protocol.SP_FE2CL_PC_EXIT{
 		IID: int32(plr.PlayerID),
 	})
 }
 
-func (plr *Player) EnterIntoViewOf(peer *cnet.Peer) {
+func (plr *Player) EnterIntoViewOf(peer *protocol.CNPeer) {
 	peer.Send(protocol.P_FE2CL_PC_NEW, protocol.SP_FE2CL_PC_NEW{
 		PCAppearanceData: plr.GetAppearanceData(),
 	})

internal/protocol/cnpeer.go

@@ -1,4 +1,4 @@
-package cnet
+package protocol
 
 import (
 	"bytes"
@@ -8,8 +8,7 @@ import (
 	"io"
 	"net"
 	"sync/atomic"
-
-	"github.com/CPunch/gopenfusion/cnet/protocol"
+	"time"
 )
 
 const (
@@ -23,8 +22,8 @@ type PacketEvent struct {
 	PktID uint32
 }
 
-// Peer is a simple wrapper for net.Conn connections to send/recv packets over the Fusionfall packet protocol.
-type Peer struct {
+// CNPeer is a simple wrapper for net.Conn connections to send/recv packets over the Fusionfall packet protocol.
+type CNPeer struct {
 	uData    interface{}
 	conn     net.Conn
 	ctx      context.Context
@@ -38,38 +37,42 @@ type Peer struct {
 	FE_key []byte
 }
 
-func NewPeer(ctx context.Context, conn net.Conn) *Peer {
-	p := &Peer{
+func GetTime() uint64 {
+	return uint64(time.Now().UnixMilli())
+}
+
+func NewCNPeer(ctx context.Context, conn net.Conn) *CNPeer {
+	p := &CNPeer{
 		conn:     conn,
 		ctx:      ctx,
 		whichKey: USE_E,
 		alive:    &atomic.Bool{},
 
-		E_key:  []byte(protocol.DEFAULT_KEY),
+		E_key:  []byte(DEFAULT_KEY),
 		FE_key: nil,
 	}
 
 	return p
 }
 
-func (peer *Peer) SetUserData(uData interface{}) {
+func (peer *CNPeer) SetUserData(uData interface{}) {
 	peer.uData = uData
 }
 
-func (peer *Peer) UserData() interface{} {
+func (peer *CNPeer) UserData() interface{} {
 	return peer.uData
 }
 
-func (peer *Peer) Send(typeID uint32, data ...interface{}) error {
+func (peer *CNPeer) Send(typeID uint32, data ...interface{}) error {
 	// grab buffer from pool
-	buf := protocol.GetBuffer()
-	defer protocol.PutBuffer(buf)
+	buf := GetBuffer()
+	defer PutBuffer(buf)
 
 	// allocate space for packet size
 	buf.Write(make([]byte, 4))
 
 	// body start
-	pkt := protocol.NewPacket(buf)
+	pkt := NewPacket(buf)
 
 	// encode type id
 	if err := pkt.Encode(typeID); err != nil {
@@ -94,7 +97,7 @@ func (peer *Peer) Send(typeID uint32, data ...interface{}) error {
 	case USE_FE:
 		key = peer.FE_key
 	}
-	protocol.EncryptData(buf.Bytes()[4:], key)
+	EncryptData(buf.Bytes()[4:], key)
 
 	// send full packet
 	// log.Printf("Sending %#v, sizeof: %d, buffer: %v", data, buf.Len(), buf.Bytes())
@@ -104,11 +107,11 @@ func (peer *Peer) Send(typeID uint32, data ...interface{}) error {
 	return nil
 }
 
-func (peer *Peer) SetActiveKey(whichKey int) {
+func (peer *CNPeer) SetActiveKey(whichKey int) {
 	peer.whichKey = whichKey
 }
 
-func (peer *Peer) Kill() {
+func (peer *CNPeer) Kill() {
 	// de-bounce: only kill if alive
 	if !peer.alive.CompareAndSwap(true, false) {
 		return
@@ -118,7 +121,7 @@ func (peer *Peer) Kill() {
 }
 
 // meant to be invoked as a goroutine
-func (peer *Peer) Handler(eRecv chan<- *PacketEvent) error {
+func (peer *CNPeer) Handler(eRecv chan<- *PacketEvent) error {
 	defer func() {
 		close(eRecv)
 		peer.Kill()
@@ -137,19 +140,19 @@ func (peer *Peer) Handler(eRecv chan<- *PacketEvent) error {
 			}
 
 			// client should never send a packet size outside of this range
-			if sz > protocol.CN_PACKET_BUFFER_SIZE || sz < 4 {
+			if sz > CN_PACKET_BUFFER_SIZE || sz < 4 {
 				return fmt.Errorf("invalid packet size: %d", sz)
 			}
 
 			// grab buffer && read packet body
-			buf := protocol.GetBuffer()
+			buf := GetBuffer()
 			if _, err := buf.ReadFrom(io.LimitReader(peer.conn, int64(sz))); err != nil {
 				return fmt.Errorf("failed to read packet body: %v", err)
 			}
 
 			// decrypt
-			protocol.DecryptData(buf.Bytes(), peer.E_key)
-			pkt := protocol.NewPacket(buf)
+			DecryptData(buf.Bytes(), peer.E_key)
+			pkt := NewPacket(buf)
 
 			// create packet && read pktID
 			var pktID uint32

internal/protocol/protocol_test.go

@@ -4,7 +4,7 @@ import (
 	"bytes"
 	"testing"
 
-	"github.com/CPunch/gopenfusion/cnet/protocol"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 	"github.com/matryer/is"
 )
 

internal/service/service.go

@@ -1,4 +1,4 @@
-package cnet
+package service
 
 import (
 	"context"
@@ -10,13 +10,13 @@ import (
 	"strconv"
 	"sync"
 
-	"github.com/CPunch/gopenfusion/cnet/protocol"
 	"github.com/CPunch/gopenfusion/config"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 )
 
-type PacketHandler func(peer *Peer, pkt protocol.Packet) error
+type PacketHandler func(peer *protocol.CNPeer, pkt protocol.Packet) error
 
-func StubbedPacket(_ *Peer, _ protocol.Packet) error {
+func StubbedPacket(_ *protocol.CNPeer, _ protocol.Packet) error {
 	return nil
 }
 
@@ -28,18 +28,18 @@ type Service struct {
 	started        chan struct{}
 	stopped        chan struct{}
 	packetHandlers map[uint32]PacketHandler
-	peers          map[chan *PacketEvent]*Peer
+	peers          map[chan *protocol.PacketEvent]*protocol.CNPeer
 	stateLock      sync.Mutex
 
 	// OnDisconnect is called when a peer disconnects from the service.
 	// uData is the stored value of the key/value pair in the peer map.
 	// It may not be set while the service is running. (eg. srvc.Start() has been called)
-	OnDisconnect func(peer *Peer)
+	OnDisconnect func(peer *protocol.CNPeer)
 
 	// OnConnect is called when a peer connects to the service.
 	// return value is used as the value in the peer map.
 	// It may not be set while the service is running. (eg. srvc.Start() has been called)
-	OnConnect func(peer *Peer)
+	OnConnect func(peer *protocol.CNPeer)
 }
 
 func RandomPort() (int, error) {
@@ -69,7 +69,7 @@ func NewService(ctx context.Context, name string, port int) *Service {
 func (srvc *Service) Reset(ctx context.Context) {
 	srvc.ctx = ctx
 	srvc.packetHandlers = make(map[uint32]PacketHandler)
-	srvc.peers = make(map[chan *PacketEvent]*Peer)
+	srvc.peers = make(map[chan *protocol.PacketEvent]*protocol.CNPeer)
 	srvc.started = make(chan struct{})
 	srvc.stopped = make(chan struct{})
 }
@@ -80,8 +80,8 @@ func (srvc *Service) AddPacketHandler(pktID uint32, handler PacketHandler) {
 }
 
 type newPeerConnection struct {
-	peer    *Peer
-	channel chan *PacketEvent
+	peer    *protocol.CNPeer
+	channel chan *protocol.PacketEvent
 }
 
 func (srvc *Service) Start() error {
@@ -112,22 +112,22 @@ func (srvc *Service) Start() error {
 		}
 
 		// create a new peer and pass it to the event loop
-		peer := NewPeer(srvc.ctx, conn)
-		eRecv := make(chan *PacketEvent)
+		peer := protocol.NewCNPeer(srvc.ctx, conn)
+		eRecv := make(chan *protocol.PacketEvent)
 		peerConnections <- newPeerConnection{channel: eRecv, peer: peer}
 		go peer.Handler(eRecv)
 	}
 }
 
-func (srvc *Service) getPeer(channel chan *PacketEvent) *Peer {
+func (srvc *Service) getPeer(channel chan *protocol.PacketEvent) *protocol.CNPeer {
 	return srvc.peers[channel]
 }
 
-func (srvc *Service) setPeer(channel chan *PacketEvent, peer *Peer) {
+func (srvc *Service) setPeer(channel chan *protocol.PacketEvent, peer *protocol.CNPeer) {
 	srvc.peers[channel] = peer
 }
 
-func (srvc *Service) removePeer(channel chan *PacketEvent) {
+func (srvc *Service) removePeer(channel chan *protocol.PacketEvent) {
 	delete(srvc.peers, channel)
 }
 
@@ -147,7 +147,7 @@ func (srvc *Service) Stopped() <-chan struct{} {
 // if f returns false, the iteration is stopped.
 // NOTE: the peer map is not locked while iterating, if you're calling this
 // outside of the service's event loop, you'll need to lock the peer map yourself.
-func (srvc *Service) RangePeers(f func(peer *Peer) bool) {
+func (srvc *Service) RangePeers(f func(peer *protocol.CNPeer) bool) {
 	for _, peer := range srvc.peers {
 		if !f(peer) {
 			break
@@ -167,7 +167,7 @@ func (srvc *Service) Unlock() {
 
 func (srvc *Service) stop() {
 	// OnDisconnect handler might need to do something important
-	srvc.RangePeers(func(peer *Peer) bool {
+	srvc.RangePeers(func(peer *protocol.CNPeer) bool {
 		peer.Kill()
 		if srvc.OnDisconnect != nil {
 			srvc.OnDisconnect(peer)
@@ -196,7 +196,7 @@ func (srvc *Service) handleEvents(peerPipe <-chan newPeerConnection) {
 		Chan: reflect.ValueOf(peerPipe),
 	})
 
-	addPoll := func(channel chan *PacketEvent) {
+	addPoll := func(channel chan *protocol.PacketEvent) {
 		poll = append(poll, reflect.SelectCase{
 			Dir:  reflect.SelectRecv,
 			Chan: reflect.ValueOf(channel),
@@ -221,7 +221,7 @@ func (srvc *Service) handleEvents(peerPipe <-chan newPeerConnection) {
 			addPoll(evnt.channel)
 			srvc.connect(evnt.channel, evnt.peer)
 		default: // peer event
-			channel := poll[chosen].Chan.Interface().(chan *PacketEvent)
+			channel := poll[chosen].Chan.Interface().(chan *protocol.PacketEvent)
 			peer := srvc.getPeer(channel)
 			if peer == nil {
 				log.Printf("Unknown peer event: %v", value)
@@ -229,7 +229,7 @@ func (srvc *Service) handleEvents(peerPipe <-chan newPeerConnection) {
 				continue
 			}
 
-			evnt, ok := value.Interface().(*PacketEvent)
+			evnt, ok := value.Interface().(*protocol.PacketEvent)
 			if !recvOK || !ok || evnt == nil {
 				// peer disconnected, remove it from our poll queue
 				removePoll(chosen)
@@ -250,7 +250,7 @@ func (srvc *Service) handleEvents(peerPipe <-chan newPeerConnection) {
 	}
 }
 
-func (srvc *Service) handlePacket(peer *Peer, typeID uint32, pkt protocol.Packet) error {
+func (srvc *Service) handlePacket(peer *protocol.CNPeer, typeID uint32, pkt protocol.Packet) error {
 	if hndlr, ok := srvc.packetHandlers[typeID]; ok {
 		// fmt.Printf("Handling packet %x\n", typeID)
 		if err := hndlr(peer, pkt); err != nil {
@@ -263,7 +263,7 @@ func (srvc *Service) handlePacket(peer *Peer, typeID uint32, pkt protocol.Packet
 	return nil
 }
 
-func (srvc *Service) disconnect(channel chan *PacketEvent, peer *Peer) {
+func (srvc *Service) disconnect(channel chan *protocol.PacketEvent, peer *protocol.CNPeer) {
 	log.Printf("Peer %p disconnected from %s\n", peer, srvc.Name)
 	if srvc.OnDisconnect != nil {
 		srvc.OnDisconnect(peer)
@@ -272,7 +272,7 @@ func (srvc *Service) disconnect(channel chan *PacketEvent, peer *Peer) {
 	srvc.removePeer(channel)
 }
 
-func (srvc *Service) connect(channel chan *PacketEvent, peer *Peer) {
+func (srvc *Service) connect(channel chan *protocol.PacketEvent, peer *protocol.CNPeer) {
 	log.Printf("New peer %p connected to %s\n", peer, srvc.Name)
 	if srvc.OnConnect != nil {
 		srvc.OnConnect(peer)

internal/service/service_test.go

@@ -0,0 +1,117 @@
+package service_test
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/CPunch/gopenfusion/internal/protocol"
+	"github.com/CPunch/gopenfusion/internal/service"
+	"github.com/matryer/is"
+)
+
+var (
+	srvcPort int
+)
+
+const (
+	timeout       = 2
+	maxDummyPeers = 5
+)
+
+func selectWithTimeout(ch <-chan struct{}, seconds int) bool {
+	select {
+	case <-ch:
+		return true
+	case <-time.After(time.Duration(seconds) * time.Second):
+		return false
+	}
+}
+
+func waitWithTimeout(wg *sync.WaitGroup, seconds int) bool {
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		wg.Wait()
+	}()
+
+	return selectWithTimeout(done, seconds)
+}
+
+func TestMain(m *testing.M) {
+	var err error
+	srvcPort, err = service.RandomPort()
+	if err != nil {
+		panic(err)
+	}
+
+	os.Exit(m.Run())
+}
+
+func TestService(t *testing.T) {
+	is := is.New(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	srvc := service.NewService(ctx, "TEST", srvcPort)
+	wg := sync.WaitGroup{}
+
+	// shutdown service when test is done
+	defer func() {
+		cancel()
+		is.True(selectWithTimeout(srvc.Stopped(), timeout)) // wait for service to stop with timeout
+	}()
+
+	// our dummy packet handler
+	srvc.AddPacketHandler(0x1234, func(peer *protocol.CNPeer, pkt protocol.Packet) error {
+		log.Printf("Received packet %#v", pkt)
+		wg.Done()
+		return nil
+	})
+
+	// wait for all dummy peers to connect and disconnect
+	wg.Add(maxDummyPeers)
+	srvc.OnConnect = func(peer *protocol.CNPeer) {
+		wg.Done()
+	}
+
+	wg.Add(maxDummyPeers)
+	srvc.OnDisconnect = func(peer *protocol.CNPeer) {
+		wg.Done()
+	}
+
+	// run service
+	go func() {
+		is.NoErr(srvc.Start()) // srvc.Start error
+	}()
+
+	is.True(selectWithTimeout(srvc.Started(), timeout)) // wait for service to start with timeout
+
+	wg.Add(maxDummyPeers * 3) // 3 wg.Done() calls per dummy peer. 2 per peer for receiving packets, 1 for Handler() exit
+	for i := 0; i < maxDummyPeers; i++ {
+		go func() {
+			// make dummy client
+			conn, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", srvcPort))
+			is.NoErr(err) // net.Dial error
+
+			peer := protocol.NewCNPeer(ctx, conn)
+			go func() {
+				defer peer.Kill()
+
+				// send dummy packets
+				for i := 0; i < 2; i++ {
+					is.NoErr(peer.Send(0x1234)) // peer.Send error
+				}
+			}()
+
+			// we wait until Handler gracefully exits (peer was killed)
+			peer.Handler(make(chan *protocol.PacketEvent))
+			wg.Done()
+		}()
+	}
+
+	is.True(waitWithTimeout(&wg, timeout)) // wait for all dummy peers to be done with timeout
+}

login/login.go

@@ -7,12 +7,10 @@ import (
 	"log"
 	"math/rand"
 
-	"github.com/CPunch/gopenfusion/cnet"
-	"github.com/CPunch/gopenfusion/cnet/protocol"
 	"github.com/CPunch/gopenfusion/config"
 	"github.com/CPunch/gopenfusion/internal/db"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 	"github.com/CPunch/gopenfusion/internal/redis"
-	"github.com/CPunch/gopenfusion/util"
 )
 
 const (
@@ -27,14 +25,14 @@ const (
 	LOGIN_UPDATED_EUALA_REQUIRED                = 9
 )
 
-func (server *LoginServer) AcceptLogin(peer *cnet.Peer, SzID string, IClientVerC int32, ISlotNum int8, data []protocol.SP_LS2CL_REP_CHAR_INFO) error {
+func (server *LoginServer) AcceptLogin(peer *protocol.CNPeer, SzID string, IClientVerC int32, ISlotNum int8, data []protocol.SP_LS2CL_REP_CHAR_INFO) error {
 	resp := protocol.SP_LS2CL_REP_LOGIN_SUCC{
 		SzID:          SzID,
 		ICharCount:    int8(len(data)),
 		ISlotNum:      ISlotNum,
 		IPaymentFlag:  1,
 		IOpenBetaFlag: 0,
-		UiSvrTime:     util.GetTime(),
+		UiSvrTime:     protocol.GetTime(),
 	}
 
 	if err := peer.Send(protocol.P_LS2CL_REP_LOGIN_SUCC, resp); err != nil {
@@ -63,7 +61,7 @@ func (server *LoginServer) AcceptLogin(peer *cnet.Peer, SzID string, IClientVerC
 	return nil
 }
 
-func (server *LoginServer) Login(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *LoginServer) Login(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var loginPkt protocol.SP_CL2LS_REQ_LOGIN
 	pkt.Decode(&loginPkt)
 
@@ -139,7 +137,7 @@ func (server *LoginServer) Login(peer *cnet.Peer, pkt protocol.Packet) error {
 	return server.AcceptLogin(peer, loginPkt.SzID, loginPkt.IClientVerC, 1, charInfo[:len(plrs)])
 }
 
-func (server *LoginServer) CheckCharacterName(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *LoginServer) CheckCharacterName(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var charPkt protocol.SP_CL2LS_REQ_CHECK_CHAR_NAME
 	pkt.Decode(&charPkt)
 
@@ -150,7 +148,7 @@ func (server *LoginServer) CheckCharacterName(peer *cnet.Peer, pkt protocol.Pack
 	})
 }
 
-func (server *LoginServer) SaveCharacterName(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *LoginServer) SaveCharacterName(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var charPkt protocol.SP_CL2LS_REQ_SAVE_CHAR_NAME
 	pkt.Decode(&charPkt)
 
@@ -203,7 +201,7 @@ func validateCharacterCreation(character *protocol.SP_CL2LS_REQ_CHAR_CREATE) boo
 	return true
 }
 
-func SendFail(peer *cnet.Peer) error {
+func SendFail(peer *protocol.CNPeer) error {
 	if err := peer.Send(protocol.P_LS2CL_REP_SHARD_SELECT_FAIL, protocol.SP_LS2CL_REP_SHARD_SELECT_FAIL{
 		IErrorCode: 2,
 	}); err != nil {
@@ -213,7 +211,7 @@ func SendFail(peer *cnet.Peer) error {
 	return nil
 }
 
-func (server *LoginServer) CharacterCreate(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *LoginServer) CharacterCreate(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var charPkt protocol.SP_CL2LS_REQ_CHAR_CREATE
 	pkt.Decode(&charPkt)
 
@@ -246,7 +244,7 @@ func (server *LoginServer) CharacterCreate(peer *cnet.Peer, pkt protocol.Packet)
 	})
 }
 
-func (server *LoginServer) CharacterDelete(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *LoginServer) CharacterDelete(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var charPkt protocol.SP_CL2LS_REQ_CHAR_DELETE
 	pkt.Decode(&charPkt)
 
@@ -265,7 +263,7 @@ func (server *LoginServer) CharacterDelete(peer *cnet.Peer, pkt protocol.Packet)
 	})
 }
 
-func (server *LoginServer) ShardSelect(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *LoginServer) ShardSelect(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var selection protocol.SP_CL2LS_REQ_CHAR_SELECT
 	pkt.Decode(&selection)
 
@@ -321,7 +319,7 @@ func (server *LoginServer) ShardSelect(peer *cnet.Peer, pkt protocol.Packet) err
 	return peer.Send(protocol.P_LS2CL_REP_SHARD_SELECT_SUCC, resp)
 }
 
-func (server *LoginServer) FinishTutorial(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *LoginServer) FinishTutorial(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var charPkt protocol.SP_CL2LS_REQ_SAVE_CHAR_TUTOR
 	pkt.Decode(&charPkt)
 

login/loginserver.go

@@ -3,20 +3,20 @@ package login
 import (
 	"context"
 
-	"github.com/CPunch/gopenfusion/cnet"
-	"github.com/CPunch/gopenfusion/cnet/protocol"
 	"github.com/CPunch/gopenfusion/internal/db"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 	"github.com/CPunch/gopenfusion/internal/redis"
+	"github.com/CPunch/gopenfusion/internal/service"
 )
 
 type LoginServer struct {
-	service    *cnet.Service
+	service    *service.Service
 	dbHndlr    *db.DBHandler
 	redisHndlr *redis.RedisHandler
 }
 
 func NewLoginServer(ctx context.Context, dbHndlr *db.DBHandler, redisHndlr *redis.RedisHandler, port int) (*LoginServer, error) {
-	srvc := cnet.NewService(ctx, "LOGIN", port)
+	srvc := service.NewService(ctx, "LOGIN", port)
 
 	server := &LoginServer{
 		service:    srvc,
@@ -30,14 +30,14 @@ func NewLoginServer(ctx context.Context, dbHndlr *db.DBHandler, redisHndlr *redi
 	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_CHAR_CREATE, server.CharacterCreate)
 	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_CHAR_SELECT, server.ShardSelect)
 	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_CHAR_DELETE, server.CharacterDelete)
-	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_SHARD_SELECT, cnet.StubbedPacket)
-	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_SHARD_LIST_INFO, cnet.StubbedPacket)
-	srvc.AddPacketHandler(protocol.P_CL2LS_CHECK_NAME_LIST, cnet.StubbedPacket)
+	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_SHARD_SELECT, service.StubbedPacket)
+	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_SHARD_LIST_INFO, service.StubbedPacket)
+	srvc.AddPacketHandler(protocol.P_CL2LS_CHECK_NAME_LIST, service.StubbedPacket)
 	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_SAVE_CHAR_TUTOR, server.FinishTutorial)
-	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_PC_EXIT_DUPLICATE, cnet.StubbedPacket)
-	srvc.AddPacketHandler(protocol.P_CL2LS_REP_LIVE_CHECK, cnet.StubbedPacket)
-	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_CHANGE_CHAR_NAME, cnet.StubbedPacket)
-	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_SERVER_SELECT, cnet.StubbedPacket)
+	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_PC_EXIT_DUPLICATE, service.StubbedPacket)
+	srvc.AddPacketHandler(protocol.P_CL2LS_REP_LIVE_CHECK, service.StubbedPacket)
+	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_CHANGE_CHAR_NAME, service.StubbedPacket)
+	srvc.AddPacketHandler(protocol.P_CL2LS_REQ_SERVER_SELECT, service.StubbedPacket)
 
 	return server, nil
 }

shard/chat.go

@@ -3,12 +3,11 @@ package shard
 import (
 	"fmt"
 
-	"github.com/CPunch/gopenfusion/cnet"
-	"github.com/CPunch/gopenfusion/cnet/protocol"
-	"github.com/CPunch/gopenfusion/shard/entity"
+	"github.com/CPunch/gopenfusion/internal/entity"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 )
 
-func (server *ShardServer) freeChat(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *ShardServer) freeChat(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var chat protocol.SP_CL2FE_REQ_SEND_FREECHAT_MESSAGE
 	pkt.Decode(&chat)
 
@@ -25,7 +24,7 @@ func (server *ShardServer) freeChat(peer *cnet.Peer, pkt protocol.Packet) error
 	})
 }
 
-func (server *ShardServer) menuChat(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *ShardServer) menuChat(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var chat protocol.SP_CL2FE_REQ_SEND_MENUCHAT_MESSAGE
 	pkt.Decode(&chat)
 
@@ -42,7 +41,7 @@ func (server *ShardServer) menuChat(peer *cnet.Peer, pkt protocol.Packet) error
 	})
 }
 
-func (server *ShardServer) emoteChat(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *ShardServer) emoteChat(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var chat protocol.SP_CL2FE_REQ_PC_AVATAR_EMOTES_CHAT
 	pkt.Decode(&chat)
 

shard/entitymanager.go

@@ -1,7 +1,7 @@
 package shard
 
 import (
-	"github.com/CPunch/gopenfusion/shard/entity"
+	"github.com/CPunch/gopenfusion/internal/entity"
 )
 
 func (server *ShardServer) addEntity(e entity.Entity) {

shard/join.go

@@ -4,14 +4,12 @@ import (
 	"fmt"
 	"log"
 
-	"github.com/CPunch/gopenfusion/cnet"
-	"github.com/CPunch/gopenfusion/cnet/protocol"
+	"github.com/CPunch/gopenfusion/internal/entity"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 	"github.com/CPunch/gopenfusion/internal/redis"
-	"github.com/CPunch/gopenfusion/shard/entity"
-	"github.com/CPunch/gopenfusion/util"
 )
 
-func (server *ShardServer) attachPlayer(peer *cnet.Peer, meta redis.LoginMetadata) (*entity.Player, error) {
+func (server *ShardServer) attachPlayer(peer *protocol.CNPeer, meta redis.LoginMetadata) (*entity.Player, error) {
 	dbPlr, err := server.dbHndlr.GetPlayer(int(meta.PlayerID))
 	if err != nil {
 		return nil, err
@@ -26,7 +24,7 @@ func (server *ShardServer) attachPlayer(peer *cnet.Peer, meta redis.LoginMetadat
 	return plr, nil
 }
 
-func (server *ShardServer) RequestEnter(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *ShardServer) RequestEnter(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var enter protocol.SP_CL2FE_REQ_PC_ENTER
 	pkt.Decode(&enter)
 
@@ -51,13 +49,13 @@ func (server *ShardServer) RequestEnter(peer *cnet.Peer, pkt protocol.Packet) er
 	resp := &protocol.SP_FE2CL_REP_PC_ENTER_SUCC{
 		IID:           int32(plr.PlayerID),
 		PCLoadData2CL: plr.ToPCLoadData2CL(),
-		UiSvrTime:     util.GetTime(),
+		UiSvrTime:     protocol.GetTime(),
 	}
 
 	// setup peer
 	peer.E_key = protocol.CreateNewKey(resp.UiSvrTime, uint64(resp.IID+1), uint64(resp.PCLoadData2CL.IFusionMatter+1))
 	peer.FE_key = loginData.FEKey
-	peer.SetActiveKey(cnet.USE_FE)
+	peer.SetActiveKey(protocol.USE_FE)
 
 	log.Printf("Player %d (AccountID %d) entered\n", resp.IID, loginData.AccountID)
 	if err := peer.Send(protocol.P_FE2CL_REP_PC_ENTER_SUCC, resp); err != nil {
@@ -67,7 +65,7 @@ func (server *ShardServer) RequestEnter(peer *cnet.Peer, pkt protocol.Packet) er
 	return nil
 }
 
-func (server *ShardServer) LoadingComplete(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *ShardServer) LoadingComplete(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var loadComplete protocol.SP_CL2FE_REQ_PC_LOADING_COMPLETE
 	pkt.Decode(&loadComplete)
 

shard/movement.go

@@ -3,10 +3,8 @@ package shard
 import (
 	"fmt"
 
-	"github.com/CPunch/gopenfusion/cnet"
-	"github.com/CPunch/gopenfusion/cnet/protocol"
-	"github.com/CPunch/gopenfusion/shard/entity"
-	"github.com/CPunch/gopenfusion/util"
+	"github.com/CPunch/gopenfusion/internal/entity"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 )
 
 func (server *ShardServer) updatePlayerPosition(plr *entity.Player, X, Y, Z, Angle int) {
@@ -17,7 +15,7 @@ func (server *ShardServer) updatePlayerPosition(plr *entity.Player, X, Y, Z, Ang
 	server.updateEntityChunk(plr, plr.GetChunkPos(), entity.MakeChunkPosition(X, Y))
 }
 
-func (server *ShardServer) playerMove(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *ShardServer) playerMove(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var move protocol.SP_CL2FE_REQ_PC_MOVE
 	pkt.Decode(&move)
 
@@ -41,11 +39,11 @@ func (server *ShardServer) playerMove(peer *cnet.Peer, pkt protocol.Packet) erro
 		CKeyValue: move.CKeyValue,
 		ISpeed:    move.ISpeed,
 		IID:       int32(plr.PlayerID),
-		ISvrTime:  util.GetTime(),
+		ISvrTime:  protocol.GetTime(),
 	})
 }
 
-func (server *ShardServer) playerStop(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *ShardServer) playerStop(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var stop protocol.SP_CL2FE_REQ_PC_STOP
 	pkt.Decode(&stop)
 
@@ -63,11 +61,11 @@ func (server *ShardServer) playerStop(peer *cnet.Peer, pkt protocol.Packet) erro
 		IY:       stop.IY,
 		IZ:       stop.IZ,
 		IID:      int32(plr.PlayerID),
-		ISvrTime: util.GetTime(),
+		ISvrTime: protocol.GetTime(),
 	})
 }
 
-func (server *ShardServer) playerJump(peer *cnet.Peer, pkt protocol.Packet) error {
+func (server *ShardServer) playerJump(peer *protocol.CNPeer, pkt protocol.Packet) error {
 	var jump protocol.SP_CL2FE_REQ_PC_JUMP
 	pkt.Decode(&jump)
 
@@ -91,6 +89,6 @@ func (server *ShardServer) playerJump(peer *cnet.Peer, pkt protocol.Packet) erro
 		CKeyValue: jump.CKeyValue,
 		ISpeed:    jump.ISpeed,
 		IID:       int32(plr.PlayerID),
-		ISvrTime:  util.GetTime(),
+		ISvrTime:  protocol.GetTime(),
 	})
 }

shard/npcloader.go

@@ -6,7 +6,7 @@ import (
 	"os"
 
 	"github.com/CPunch/gopenfusion/config"
-	"github.com/CPunch/gopenfusion/shard/entity"
+	"github.com/CPunch/gopenfusion/internal/entity"
 )
 
 type NPCData struct {

shard/shardserver.go

@@ -3,25 +3,25 @@ package shard
 import (
 	"context"
 
-	"github.com/CPunch/gopenfusion/cnet"
-	"github.com/CPunch/gopenfusion/cnet/protocol"
 	"github.com/CPunch/gopenfusion/config"
 	"github.com/CPunch/gopenfusion/internal/db"
+	"github.com/CPunch/gopenfusion/internal/entity"
+	"github.com/CPunch/gopenfusion/internal/protocol"
 	"github.com/CPunch/gopenfusion/internal/redis"
-	"github.com/CPunch/gopenfusion/shard/entity"
+	"github.com/CPunch/gopenfusion/internal/service"
 )
 
-type PacketHandler func(peer *cnet.Peer, pkt protocol.Packet) error
+type PacketHandler func(peer *protocol.CNPeer, pkt protocol.Packet) error
 
 type ShardServer struct {
-	service    *cnet.Service
+	service    *service.Service
 	dbHndlr    *db.DBHandler
 	redisHndlr *redis.RedisHandler
 	chunks     map[entity.ChunkPosition]*entity.Chunk
 }
 
 func NewShardServer(ctx context.Context, dbHndlr *db.DBHandler, redisHndlr *redis.RedisHandler, port int) (*ShardServer, error) {
-	srvc := cnet.NewService(ctx, "SHARD", port)
+	srvc := service.NewService(ctx, "SHARD", port)
 
 	server := &ShardServer{
 		service:    srvc,
@@ -55,7 +55,7 @@ func (server *ShardServer) Start() {
 	server.service.Start()
 }
 
-func (server *ShardServer) onDisconnect(peer *cnet.Peer) {
+func (server *ShardServer) onDisconnect(peer *protocol.CNPeer) {
 	// remove from chunks
 	plr, ok := peer.UserData().(*entity.Player)
 	if ok && plr != nil {
@@ -63,6 +63,6 @@ func (server *ShardServer) onDisconnect(peer *cnet.Peer) {
 	}
 }
 
-func (server *ShardServer) onConnect(peer *cnet.Peer) {
+func (server *ShardServer) onConnect(peer *protocol.CNPeer) {
 
 }

util/util.go

@@ -1,29 +0,0 @@
-package util
-
-import (
-	"sync"
-	"time"
-)
-
-func GetTime() uint64 {
-	return uint64(time.Now().UnixMilli())
-}
-
-func SelectWithTimeout(ch <-chan struct{}, timeout time.Duration) bool {
-	select {
-	case <-ch:
-		return true
-	case <-time.After(timeout):
-		return false
-	}
-}
-
-func WaitWithTimeout(wg *sync.WaitGroup, timeout time.Duration) bool {
-	done := make(chan struct{})
-	go func() {
-		defer close(done)
-		wg.Wait()
-	}()
-
-	return SelectWithTimeout(done, timeout)
-}

util/util_test.go

@@ -1,35 +0,0 @@
-package util_test
-
-import (
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/CPunch/gopenfusion/util"
-	"github.com/matryer/is"
-)
-
-func TestWaitWithTimeout(t *testing.T) {
-	is := is.New(t)
-	wg := &sync.WaitGroup{}
-	go func() {
-		time.Sleep(1 * time.Second)
-		wg.Done()
-	}()
-
-	wg.Add(1)
-	is.True(!util.WaitWithTimeout(wg, 500*time.Millisecond)) // timeout should occur
-	is.True(util.WaitWithTimeout(wg, 750*time.Millisecond))  // timeout shouldn't occur
-}
-
-func TestSelectWithTimeout(t *testing.T) {
-	is := is.New(t)
-	ch := make(chan struct{})
-	go func() {
-		time.Sleep(1 * time.Second)
-		close(ch)
-	}()
-
-	is.True(!util.SelectWithTimeout(ch, 500*time.Millisecond)) // timeout should occur
-	is.True(util.SelectWithTimeout(ch, 750*time.Millisecond))  // timeout shouldn't occur
-}