server: split login/ and shard/

shard/join.go

@@ -0,0 +1,64 @@
+package shard
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/CPunch/gopenfusion/core"
+	"github.com/CPunch/gopenfusion/core/protocol"
+)
+
+func (server *ShardServer) RequestEnter(peer *protocol.CNPeer, pkt protocol.Packet) error {
+	var enter protocol.SP_CL2FE_REQ_PC_ENTER
+	pkt.Decode(&enter)
+
+	loginData, err := server.CheckLogin(enter.IEnterSerialKey)
+	if err != nil {
+		// the error codes for P_FE2CL_REP_PC_ENTER_FAIL aren't referenced in the client :(
+		peer.Send(protocol.P_FE2CL_REP_PC_ENTER_FAIL, protocol.SP_FE2CL_REP_PC_ENTER_FAIL{})
+		return err
+	}
+
+	// attach player
+	var resp *protocol.SP_FE2CL_REP_PC_ENTER_SUCC
+	if err := server.UpdatePlayer(peer, func(old *core.Player) (*core.Player, error) {
+		if old != nil { // resending a shard enter packet?
+			return nil, fmt.Errorf("resent enter packet!")
+		}
+
+		plr, err := server.dbHndlr.GetPlayer(int(loginData.PlayerID))
+		if err != nil {
+			return nil, err
+		}
+
+		resp = &protocol.SP_FE2CL_REP_PC_ENTER_SUCC{
+			IID:           int32(plr.PlayerID),
+			PCLoadData2CL: plr.ToPCLoadData2CL(),
+			UiSvrTime:     uint64(time.Now().Unix()),
+		}
+
+		return plr, nil
+	}); err != nil {
+		peer.Send(protocol.P_FE2CL_REP_PC_ENTER_FAIL, protocol.SP_FE2CL_REP_PC_ENTER_FAIL{})
+		return err
+	}
+
+	// setup key
+	peer.E_key = protocol.CreateNewKey(resp.UiSvrTime, uint64(resp.IID+1), uint64(resp.PCLoadData2CL.IFusionMatter+1))
+	peer.FE_key = loginData.FEKey
+	peer.SetActiveKey(protocol.USE_FE)
+
+	return peer.Send(protocol.P_FE2CL_REP_PC_ENTER_SUCC, resp)
+}
+
+func (server *ShardServer) LoadingComplete(peer *protocol.CNPeer, pkt protocol.Packet) error {
+	var loadComplete protocol.SP_CL2FE_REQ_PC_LOADING_COMPLETE
+	pkt.Decode(&loadComplete)
+
+	plr := server.LoadPlayer(peer)
+	if plr == nil {
+		return fmt.Errorf("peer has no player attached!")
+	}
+
+	return peer.Send(protocol.P_FE2CL_REP_PC_LOADING_COMPLETE_SUCC, protocol.SP_FE2CL_REP_PC_LOADING_COMPLETE_SUCC{IPC_ID: int32(plr.PlayerID)})
+}

shard/shardServer.go

@@ -0,0 +1,171 @@
+package shard
+
+import (
+	"fmt"
+	"log"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/CPunch/gopenfusion/core"
+	"github.com/CPunch/gopenfusion/core/db"
+	"github.com/CPunch/gopenfusion/core/protocol"
+)
+
+type LoginMetadata struct {
+	FEKey     []byte
+	Timestamp time.Time
+	PlayerID  int32
+}
+
+type PacketHandler func(peer *protocol.CNPeer, pkt protocol.Packet) error
+
+func stubbedPacket(_ *protocol.CNPeer, _ protocol.Packet) error { /* stubbed */ return nil }
+
+type ShardServer struct {
+	listener           net.Listener
+	port               int
+	dbHndlr            *db.DBHandler
+	packetHandlers     map[uint32]PacketHandler
+	loginMetadataQueue sync.Map // [int64]*LoginMetadata w/ int64 = serialKey
+	peersLock          sync.Mutex
+	peers              sync.Map // [*protocol.CNPeer]core.Player
+}
+
+func NewShardServer(dbHndlr *db.DBHandler, port int) (*ShardServer, error) {
+	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
+	if err != nil {
+		return nil, err
+	}
+
+	server := &ShardServer{
+		listener:       listener,
+		port:           port,
+		dbHndlr:        dbHndlr,
+		packetHandlers: make(map[uint32]PacketHandler),
+	}
+
+	server.packetHandlers = map[uint32]PacketHandler{
+		protocol.P_CL2FE_REQ_PC_ENTER:            server.RequestEnter,
+		protocol.P_CL2FE_REQ_PC_LOADING_COMPLETE: server.LoadingComplete,
+	}
+
+	return server, nil
+}
+
+func (server *ShardServer) RegisterPacketHandler(typeID uint32, hndlr PacketHandler) {
+	server.packetHandlers[typeID] = hndlr
+}
+
+func (server *ShardServer) Start() {
+	log.Printf("Server hosted on 127.0.0.1:%d\n", server.port)
+
+	for {
+		conn, err := server.listener.Accept()
+		if err != nil {
+			log.Println("Connection error: ", err)
+			return
+		}
+
+		client := protocol.NewCNPeer(server, conn)
+		server.Connect(client)
+		go client.Handler()
+	}
+}
+
+func (server *ShardServer) GetPort() int {
+	return server.port
+}
+
+func (server *ShardServer) HandlePacket(peer *protocol.CNPeer, typeID uint32, pkt protocol.Packet) error {
+	if hndlr, ok := server.packetHandlers[typeID]; ok {
+		if err := hndlr(peer, pkt); err != nil {
+			return err
+		}
+	} else {
+		log.Printf("[WARN] invalid packet ID: %x\n", typeID)
+	}
+
+	return nil
+}
+
+func (server *ShardServer) Disconnect(peer *protocol.CNPeer) {
+	log.Printf("Peer %p disconnected from SHARD\n", peer)
+	server.peers.Delete(peer)
+}
+
+func (server *ShardServer) Connect(peer *protocol.CNPeer) {
+	log.Printf("New peer %p connected to SHARD\n", peer)
+	server.peers.Store(peer, nil)
+}
+
+func (server *ShardServer) LoadPlayer(peer *protocol.CNPeer) *core.Player {
+	val, ok := server.peers.Load(peer)
+	if !ok {
+		return nil
+	}
+
+	plr, ok := val.(*core.Player)
+	if !ok {
+		return nil
+	}
+
+	return plr
+}
+
+// UpdatePlayer locks the peers map, and calls the provided callback. The returned new pointer will be stored, however if an error returns it will be passed back.
+// Since it is UNSAFE to write to the returned pointer from LoadPlayer, this wrapper is for the cases that state in the player struct needs to be updated.
+// TODO: maybe LoadPlayer should return a player by value instead?
+// The pointers new and old may be the same if you are just updating struct fields.
+func (server *ShardServer) UpdatePlayer(peer *protocol.CNPeer, f func(old *core.Player) (new *core.Player, err error)) error {
+	server.peersLock.Lock()
+	defer server.peersLock.Unlock()
+
+	// on fail, the player should not be stored
+	new, err := f(server.LoadPlayer(peer))
+	if err != nil {
+		return err
+	}
+
+	server.storePlayer(peer, new)
+	return nil
+}
+
+func (server *ShardServer) storePlayer(peer *protocol.CNPeer, player *core.Player) {
+	server.peers.Store(peer, player)
+}
+
+// Simple wrapper for server.peers.Range, if f returns false the iteration is stopped.
+func (server *ShardServer) RangePeers(f func(peer *protocol.CNPeer, player *core.Player) bool) {
+	server.peers.Range(func(key, value any) bool { // simple wrapper to cast the datatypes
+		peer, ok := key.(*protocol.CNPeer)
+		if !ok { // this should never happen
+			panic(fmt.Errorf("ShardServer.peers has an invalid key: peers[%#v] = %#v", key, value))
+		}
+
+		player, ok := value.(*core.Player)
+		if !ok { // this should also never happen
+			panic(fmt.Errorf("ShardServer.peers has an invalid value: peers[%#v] = %#v", key, value))
+		}
+
+		return f(peer, player)
+	})
+}
+
+func (server *ShardServer) QueueLogin(serialKey int64, meta *LoginMetadata) {
+	server.loginMetadataQueue.Store(serialKey, meta)
+}
+
+func (server *ShardServer) CheckLogin(serialKey int64) (*LoginMetadata, error) {
+	value, ok := server.loginMetadataQueue.Load(serialKey)
+	if !ok {
+		return nil, fmt.Errorf("serialKey %x is not valid!", serialKey)
+	}
+
+	meta, ok := value.(*LoginMetadata)
+	if !ok { // should never happen
+		panic(fmt.Errorf("ShardServer.loginMetadataQueue has an invalid value: loginMetadataQueue[%x] = %#v", serialKey, value))
+	}
+
+	return meta, nil
+}