holy refactor

started out as me making a service abstraction..

- db.Player exists again, and entity.Player uses it as an embedded struct
- chunk.ForEachEntity() lets you add/remove entities during iteration now
- removed account related fields from CNPeer
- protocol/pool has been merged with protocol.
use protocol.GetBuffer() and protocol.PutBuffer().
- new protocol/internal/service!
service.Service is an abstraction layer to handle multiple CNPeer*
connections and allows you to associate each with an interface{} uData.
In the future it might also handle a task queue for jobs that
modify/interact with the player's uData, called from service.handleEvents()
- PacketHandler callback type has a new param! uData is passed as well now
- much of loginserver/shardserver is now handled by the shared service
abstraction
- SHARD: NPC_ENTER packets are now sent on player loading complete
rather than on enter.

internal/db/players.go

@@ -4,11 +4,41 @@ import (
 	"database/sql"
 
 	"github.com/CPunch/gopenfusion/config"
-	"github.com/CPunch/gopenfusion/internal/entity"
 	"github.com/CPunch/gopenfusion/internal/protocol"
 	"github.com/blockloop/scan"
 )
 
+type Player struct {
+	PlayerID           int
+	AccountID          int
+	AccountLevel       int
+	Slot               int
+	PCStyle            protocol.SPCStyle
+	PCStyle2           protocol.SPCStyle2
+	EquippedNanos      [3]int
+	Nanos              [config.NANO_COUNT]protocol.SNano
+	Equip              [config.AEQUIP_COUNT]protocol.SItemBase
+	Inven              [config.AINVEN_COUNT]protocol.SItemBase
+	Bank               [config.ABANK_COUNT]protocol.SItemBase
+	SkywayLocationFlag []byte
+	FirstUseFlag       []byte
+	Quests             []byte
+	HP                 int
+	Level              int
+	Taros              int
+	FusionMatter       int
+	Mentor             int
+	X, Y, Z            int
+	Angle              int
+	BatteryN           int
+	BatteryW           int
+	WarpLocationFlag   int
+	ActiveNanoSlotNum  int
+	Fatigue            int
+	CurrentMissionID   int
+	IPCState           int8
+}
+
 // returns PlayerID, error
 func (db *DBHandler) NewPlayer(AccountID int, FirstName, LastName string, slot int) (int, error) {
 	nameCheck := 1 // for now, we approve all names
@@ -121,8 +151,8 @@ const (
 	INNER JOIN Accounts as acc ON p.AccountID = acc.AccountID `
 )
 
-func (db *DBHandler) readPlayer(rows *sql.Rows) (*entity.Player, error) {
-	plr := entity.Player{ActiveNanoSlotNum: 0}
+func (db *DBHandler) readPlayer(rows *sql.Rows) (*Player, error) {
+	plr := Player{ActiveNanoSlotNum: 0}
 
 	if err := rows.Scan(
 		&plr.PlayerID, &plr.AccountID, &plr.Slot, &plr.PCStyle.SzFirstName, &plr.PCStyle.SzLastName,
@@ -162,13 +192,13 @@ func (db *DBHandler) readPlayer(rows *sql.Rows) (*entity.Player, error) {
 	return &plr, nil
 }
 
-func (db *DBHandler) GetPlayer(PlayerID int) (*entity.Player, error) {
+func (db *DBHandler) GetPlayer(PlayerID int) (*Player, error) {
 	rows, err := db.Query(QUERY_PLAYERS+"WHERE p.PlayerID = $1", PlayerID)
 	if err != nil {
 		return nil, err
 	}
 
-	var plr *entity.Player
+	var plr *Player
 	for rows.Next() {
 		plr, err = db.readPlayer(rows)
 		if err != nil {
@@ -179,13 +209,13 @@ func (db *DBHandler) GetPlayer(PlayerID int) (*entity.Player, error) {
 	return plr, nil
 }
 
-func (db *DBHandler) GetPlayers(AccountID int) ([]entity.Player, error) {
+func (db *DBHandler) GetPlayers(AccountID int) ([]Player, error) {
 	rows, err := db.Query(QUERY_PLAYERS+"WHERE p.AccountID = $1", AccountID)
 	if err != nil {
 		return nil, err
 	}
 
-	var plrs []entity.Player
+	var plrs []Player
 	for rows.Next() {
 		plr, err := db.readPlayer(rows)
 		if err != nil {

internal/entity/chunk.go

@@ -38,11 +38,17 @@ func (c *Chunk) SendPacket(typeID uint32, pkt ...interface{}) {
 }
 
 // calls f for each entity in this chunk, if f returns true, stop iterating
+// f can safely add/remove entities from the chunk
 func (c *Chunk) ForEachEntity(f func(entity Entity) bool) {
+	// copy entities to avoid locking for the entire iteration
+	entities := make(map[Entity]struct{})
 	c.lock.Lock()
-	defer c.lock.Unlock()
-
 	for entity := range c.entities {
+		entities[entity] = struct{}{}
+	}
+	c.lock.Unlock()
+
+	for entity := range entities {
 		if f(entity) {
 			break
 		}

internal/entity/player.go

@@ -1,41 +1,22 @@
 package entity
 
 import (
-	"github.com/CPunch/gopenfusion/config"
+	"github.com/CPunch/gopenfusion/internal/db"
 	"github.com/CPunch/gopenfusion/internal/protocol"
 )
 
 type Player struct {
-	Peer               *protocol.CNPeer
-	Chunk              ChunkPosition
-	PlayerID           int
-	AccountID          int
-	AccountLevel       int
-	Slot               int
-	PCStyle            protocol.SPCStyle
-	PCStyle2           protocol.SPCStyle2
-	EquippedNanos      [3]int
-	Nanos              [config.NANO_COUNT]protocol.SNano
-	Equip              [config.AEQUIP_COUNT]protocol.SItemBase
-	Inven              [config.AINVEN_COUNT]protocol.SItemBase
-	Bank               [config.ABANK_COUNT]protocol.SItemBase
-	SkywayLocationFlag []byte
-	FirstUseFlag       []byte
-	Quests             []byte
-	HP                 int
-	Level              int
-	Taros              int
-	FusionMatter       int
-	Mentor             int
-	X, Y, Z            int
-	Angle              int
-	BatteryN           int
-	BatteryW           int
-	WarpLocationFlag   int
-	ActiveNanoSlotNum  int
-	Fatigue            int
-	CurrentMissionID   int
-	IPCState           int8
+	db.Player
+	Peer  *protocol.CNPeer
+	Chunk ChunkPosition
+}
+
+func NewPlayer(peer *protocol.CNPeer, player *db.Player) *Player {
+	return &Player{
+		Player: *player,
+		Peer:   peer,
+		Chunk:  MakeChunkPosition(player.X, player.Y),
+	}
 }
 
 // ==================== Entity interface ====================

internal/protocol/cnpeer.go

@@ -6,9 +6,8 @@ import (
 	"io"
 	"log"
 	"net"
+	"sync/atomic"
 	"time"
-
-	"github.com/CPunch/gopenfusion/internal/protocol/pool"
 )
 
 const (
@@ -18,15 +17,16 @@ const (
 
 // CNPeer is a simple wrapper for net.Conn connections to send/recv packets over the Fusionfall packet protocol.
 type CNPeer struct {
-	conn      net.Conn
-	eRecv     chan *Event
-	SzID      string
-	E_key     []byte
-	FE_key    []byte
-	AccountID int
-	PlayerID  int32
-	whichKey  int
-	alive     bool
+	conn     net.Conn
+	eRecv    chan *Event
+	whichKey int
+	alive    *atomic.Bool
+
+	// May not be set while Send() or Handler() are concurrently running.
+	E_key []byte
+
+	// May not be set while Send() or Handler() are concurrently running.
+	FE_key []byte
 }
 
 func GetTime() uint64 {
@@ -34,22 +34,23 @@ func GetTime() uint64 {
 }
 
 func NewCNPeer(eRecv chan *Event, conn net.Conn) *CNPeer {
-	return &CNPeer{
-		conn:      conn,
-		eRecv:     eRecv,
-		SzID:      "",
-		E_key:     []byte(DEFAULT_KEY),
-		FE_key:    nil,
-		AccountID: -1,
-		whichKey:  USE_E,
-		alive:     true,
+	p := &CNPeer{
+		conn:     conn,
+		eRecv:    eRecv,
+		whichKey: USE_E,
+		alive:    &atomic.Bool{},
+
+		E_key:  []byte(DEFAULT_KEY),
+		FE_key: nil,
 	}
+
+	return p
 }
 
 func (peer *CNPeer) Send(typeID uint32, data ...interface{}) error {
 	// grab buffer from pool
-	buf := pool.Get()
-	defer pool.Put(buf)
+	buf := GetBuffer()
+	defer PutBuffer(buf)
 
 	// allocate space for packet size
 	buf.Write(make([]byte, 4))
@@ -73,12 +74,14 @@ func (peer *CNPeer) Send(typeID uint32, data ...interface{}) error {
 	binary.LittleEndian.PutUint32(buf.Bytes()[:4], uint32(buf.Len()-4))
 
 	// encrypt body
+	var key []byte
 	switch peer.whichKey {
 	case USE_E:
-		EncryptData(buf.Bytes()[4:], peer.E_key)
+		key = peer.E_key
 	case USE_FE:
-		EncryptData(buf.Bytes()[4:], peer.FE_key)
+		key = peer.FE_key
 	}
+	EncryptData(buf.Bytes()[4:], key)
 
 	// send full packet
 	log.Printf("Sending %#v, sizeof: %d, buffer: %v", data, buf.Len(), buf.Bytes())
@@ -94,11 +97,12 @@ func (peer *CNPeer) SetActiveKey(whichKey int) {
 
 func (peer *CNPeer) Kill() {
 	log.Printf("Killing peer %p", peer)
-	if !peer.alive {
+
+	if !peer.alive.Load() {
 		return
 	}
+	peer.alive.Store(false)
 
-	peer.alive = false
 	peer.conn.Close()
 	peer.eRecv <- &Event{Type: EVENT_CLIENT_DISCONNECT, Peer: peer}
 }
@@ -107,6 +111,7 @@ func (peer *CNPeer) Kill() {
 func (peer *CNPeer) Handler() {
 	defer peer.Kill()
 
+	peer.alive.Store(true)
 	for {
 		// read packet size, the goroutine spends most of it's time parked here
 		var sz uint32
@@ -123,7 +128,7 @@ func (peer *CNPeer) Handler() {
 
 		// grab buffer && read packet body
 		if err := func() error {
-			buf := pool.Get()
+			buf := GetBuffer()
 			if _, err := buf.ReadFrom(io.LimitReader(peer.conn, int64(sz))); err != nil {
 				return fmt.Errorf("failed to read packet body! %v", err)
 			}

internal/protocol/pool.go

@@ -1,4 +1,4 @@
-package pool
+package protocol
 
 import (
 	"bytes"
@@ -9,11 +9,13 @@ var allocator = &sync.Pool{
 	New: func() any { return new(bytes.Buffer) },
 }
 
-func Get() *bytes.Buffer {
+// grabs a *bytes.Buffer from the pool
+func GetBuffer() *bytes.Buffer {
 	return allocator.Get().(*bytes.Buffer)
 }
 
-func Put(buf *bytes.Buffer) {
+// returns a *bytes.Buffer to the pool
+func PutBuffer(buf *bytes.Buffer) {
 	buf.Reset()
 	allocator.Put(buf)
 }

internal/service/service.go

@@ -0,0 +1,138 @@
+package service
+
+import (
+	"fmt"
+	"log"
+	"net"
+	"sync"
+
+	"github.com/CPunch/gopenfusion/config"
+	"github.com/CPunch/gopenfusion/internal/protocol"
+)
+
+type PacketHandler func(peer *protocol.CNPeer, uData interface{}, pkt protocol.Packet) error
+
+func StubbedPacket(_ *protocol.CNPeer, _ interface{}, _ protocol.Packet) error {
+	return nil
+}
+
+type Service struct {
+	listener       net.Listener
+	port           int
+	Name           string
+	eRecv          chan *protocol.Event
+	packetHandlers map[uint32]PacketHandler
+	peers          *sync.Map
+
+	// 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 *protocol.CNPeer, uData interface{})
+
+	// 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 *protocol.CNPeer) (uData interface{})
+}
+
+func NewService(name string, port int) (*Service, error) {
+	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
+	if err != nil {
+		return nil, err
+	}
+
+	service := &Service{
+		listener:       listener,
+		port:           port,
+		Name:           name,
+		eRecv:          make(chan *protocol.Event),
+		packetHandlers: make(map[uint32]PacketHandler),
+		peers:          &sync.Map{},
+	}
+
+	return service, nil
+}
+
+// may not be called while the service is running (eg. srvc.Start() has been called)
+func (service *Service) AddPacketHandler(pktID uint32, handler PacketHandler) {
+	service.packetHandlers[pktID] = handler
+}
+
+func (service *Service) Start() {
+	log.Printf("%s service hosted on %s:%d\n", service.Name, config.GetAnnounceIP(), service.port)
+
+	go service.handleEvents()
+	for {
+		conn, err := service.listener.Accept()
+		if err != nil {
+			log.Println("Connection error: ", err)
+			return
+		}
+
+		peer := protocol.NewCNPeer(service.eRecv, conn)
+		service.connect(peer)
+	}
+}
+
+func (service *Service) handleEvents() {
+	for event := range service.eRecv {
+		switch event.Type {
+		case protocol.EVENT_CLIENT_DISCONNECT:
+			service.disconnect(event.Peer)
+		case protocol.EVENT_CLIENT_PACKET:
+			if err := service.handlePacket(event.Peer, event.PktID, protocol.NewPacket(event.Pkt)); err != nil {
+				log.Printf("Error handling packet: %v", err)
+				event.Peer.Kill()
+			}
+
+			// the packet buffer is given to us by the event, so we'll need to make sure to return it to the pool
+			protocol.PutBuffer(event.Pkt)
+		}
+	}
+}
+
+func (service *Service) handlePacket(peer *protocol.CNPeer, typeID uint32, pkt protocol.Packet) error {
+	uData, _ := service.peers.Load(peer)
+	if hndlr, ok := service.packetHandlers[typeID]; ok {
+		if err := hndlr(peer, uData, pkt); err != nil {
+			return err
+		}
+	} else {
+		log.Printf("[WARN] unknown packet ID: %x\n", typeID)
+	}
+
+	return nil
+}
+
+func (service *Service) disconnect(peer *protocol.CNPeer) {
+	if service.OnDisconnect != nil {
+		uData, _ := service.peers.Load(peer)
+		service.OnDisconnect(peer, uData)
+	}
+
+	log.Printf("Peer %p disconnected from %s\n", peer, service.Name)
+	service.peers.Delete(peer)
+}
+
+func (service *Service) connect(peer *protocol.CNPeer) {
+	// default uData to nil, but if the service has an OnConnect
+	// handler, use the result from that
+	uData := interface{}(nil)
+	if service.OnConnect != nil {
+		uData = service.OnConnect(peer)
+	}
+
+	log.Printf("New peer %p connected to %s\n", peer, service.Name)
+	service.peers.Store(peer, uData)
+	go peer.Handler()
+}
+
+func (service *Service) SetPeerData(peer *protocol.CNPeer, uData interface{}) {
+	service.peers.Store(peer, uData)
+}
+
+func (service *Service) RangePeers(f func(peer *protocol.CNPeer, uData interface{}) bool) {
+	service.peers.Range(func(key, value any) bool {
+		return f(key.(*protocol.CNPeer), value)
+	})
+}