Connection manager for cleaning up connections (#404)

* Connection manager implementation to disconnect timed out users.

* Connection manager implementation to disconnect timed out users.

d2networking/d2client/d2localclient/local_client_connection.go

@@ -3,6 +3,7 @@ package d2localclient
 import (
 	"github.com/OpenDiablo2/OpenDiablo2/d2game/d2player"
 	"github.com/OpenDiablo2/OpenDiablo2/d2networking"
+	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2client/d2clientconnectiontype"
 	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2netpacket"
 	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2server"
 	uuid "github.com/satori/go.uuid"
@@ -19,8 +20,8 @@ func (l LocalClientConnection) GetUniqueId() string {
 	return l.uniqueId
 }
 
-func (l LocalClientConnection) GetConnectionType() string {
-	return "Local Client"
+func (l LocalClientConnection) GetConnectionType() d2clientconnectiontype.ClientConnectionType {
+	return d2clientconnectiontype.Local
 }
 
 func (l *LocalClientConnection) SendPacketToClient(packet d2netpacket.NetPacket) error {
@@ -45,6 +46,7 @@ func (l *LocalClientConnection) Open(connectionString string, saveFilePath strin
 }
 
 func (l *LocalClientConnection) Close() error {
+	l.SendPacketToServer(d2netpacket.CreateServerClosedPacket())
 	d2server.OnClientDisconnected(l)
 	d2server.Destroy()
 	return nil

d2networking/d2client/d2remoteclient/remote_client_connection.go

@@ -5,6 +5,7 @@ import (
 	"compress/gzip"
 	"encoding/json"
 	"fmt"
+	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2client/d2clientconnectiontype"
 	"io"
 	"log"
 	"net"
@@ -29,8 +30,8 @@ func (l RemoteClientConnection) GetUniqueId() string {
 	return l.uniqueId
 }
 
-func (l RemoteClientConnection) GetConnectionType() string {
-	return "Remote Client"
+func (l RemoteClientConnection) GetConnectionType() d2clientconnectiontype.ClientConnectionType {
+	return d2clientconnectiontype.LANClient
 }
 
 func (l *RemoteClientConnection) SendPacketToClient(packet d2netpacket.NetPacket) error { // WHAT IS THIS
@@ -76,7 +77,8 @@ func (l *RemoteClientConnection) Open(connectionString string, saveFilePath stri
 
 func (l *RemoteClientConnection) Close() error {
 	l.active = false
-	// TODO: Disconnect from the server - send a disconnect packet
+	l.SendPacketToServer(d2netpacket.CreatePlayerDisconnectRequestPacket(l.GetUniqueId()))
+
 	return nil
 }
 
@@ -147,6 +149,12 @@ func (l *RemoteClientConnection) serverListener() {
 				PacketType: packetType,
 				PacketData: packet,
 			})
+		case d2netpackettype.Ping:
+			l.SendPacketToServer(d2netpacket.CreatePongPacket(l.uniqueId))
+		case d2netpackettype.PlayerDisconnectionNotification:
+			var packet d2netpacket.PlayerDisconnectRequestPacket
+			json.Unmarshal([]byte(stringData), &packet)
+			log.Printf("Received disconnect: %s", packet.Id)
 		default:
 			fmt.Printf("Unknown packet type %d\n", packetType)
 		}

d2networking/d2client/game_client.go

@@ -3,6 +3,7 @@ package d2client
 import (
 	"fmt"
 	"log"
+	"os"
 
 	"github.com/OpenDiablo2/OpenDiablo2/d2core/d2map/d2mapgen"
 
@@ -102,6 +103,12 @@ func (g *GameClient) OnPacketReceived(packet d2netpacket.NetPacket) error {
 				player.AnimatedComposite.SetAnimationMode(player.GetAnimationMode().String())
 			})
 		}
+	case d2netpackettype.Ping:
+		g.clientConnection.SendPacketToServer(d2netpacket.CreatePongPacket(g.PlayerId))
+	case d2netpackettype.ServerClosed:
+		// TODO: Need to be tied into a character save and exit
+		log.Print("Server has been closed")
+		os.Exit(0)
 	default:
 		log.Fatalf("Invalid packet type: %d", packet.PacketType)
 	}

d2networking/d2netpacket/d2netpackettype/message_type.go

@@ -7,9 +7,13 @@ type NetPacketType uint32
 //          Also note that the packet id is a byte, so if we use more than 256
 //          of these then we are doing something very wrong.
 const (
-	UpdateServerInfo        NetPacketType = iota
-	GenerateMap                           // Sent by the server to generate a map
-	AddPlayer                             // Server sends to the client to add a player
-	MovePlayer                            // Sent to the client or server to indicate player movement
-	PlayerConnectionRequest               // Client sends to server to request a connection
+	UpdateServerInfo                NetPacketType = iota
+	GenerateMap                                   // Sent by the server to generate a map
+	AddPlayer                                     // Server sends to the client to add a player
+	MovePlayer                                    // Sent to the client or server to indicate player movement
+	PlayerConnectionRequest                       // Client sends to server to request a connection
+	PlayerDisconnectionNotification               // Client notifies the server that it is disconnecting
+	Ping                                          // Ping message type
+	Pong                                          // Pong message type
+	ServerClosed                                  // Local host has closed the server
 )

d2networking/d2netpacket/packet_ping.go

@@ -0,0 +1,19 @@
+package d2netpacket
+
+import (
+	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2netpacket/d2netpackettype"
+	"time"
+)
+
+type PingPacket struct {
+	TS time.Time `json:"ts"`
+}
+
+func CreatePingPacket() NetPacket {
+	return NetPacket{
+		PacketType: d2netpackettype.Ping,
+		PacketData: PingPacket{
+			TS: time.Now(),
+		},
+	}
+}

d2networking/d2netpacket/packet_player_disconnect_request.go

@@ -0,0 +1,20 @@
+package d2netpacket
+
+import (
+	"github.com/OpenDiablo2/OpenDiablo2/d2game/d2player"
+	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2netpacket/d2netpackettype"
+)
+
+type PlayerDisconnectRequestPacket struct {
+	Id          string                `json:"id"`
+	PlayerState *d2player.PlayerState `json:"gameState"`
+}
+
+func CreatePlayerDisconnectRequestPacket(id string) NetPacket {
+	return NetPacket{
+		PacketType: d2netpackettype.PlayerDisconnectionNotification,
+		PacketData: PlayerDisconnectRequestPacket{
+			Id: id,
+		},
+	}
+}

d2networking/d2netpacket/packet_pong.go

@@ -0,0 +1,21 @@
+package d2netpacket
+
+import (
+	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2netpacket/d2netpackettype"
+	"time"
+)
+
+type PongPacket struct {
+	ID string    `json:"id"`
+	TS time.Time `json:"ts"`
+}
+
+func CreatePongPacket(id string) NetPacket {
+	return NetPacket{
+		PacketType: d2netpackettype.Pong,
+		PacketData: PongPacket{
+			ID: id,
+			TS: time.Now(),
+		},
+	}
+}

d2networking/d2netpacket/packet_server_closed.go

@@ -0,0 +1,19 @@
+package d2netpacket
+
+import (
+	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2netpacket/d2netpackettype"
+	"time"
+)
+
+type ServerClosedPacket struct {
+	TS time.Time `json:"ts"`
+}
+
+func CreateServerClosedPacket() NetPacket {
+	return NetPacket{
+		PacketType: d2netpackettype.ServerClosed,
+		PacketData: ServerClosedPacket{
+			TS: time.Now(),
+		},
+	}
+}

d2networking/d2server/client_connection.go

@@ -2,12 +2,13 @@ package d2server
 
 import (
 	"github.com/OpenDiablo2/OpenDiablo2/d2game/d2player"
+	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2client/d2clientconnectiontype"
 	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2netpacket"
 )
 
 type ClientConnection interface {
 	GetUniqueId() string
-	GetConnectionType() string
+	GetConnectionType() d2clientconnectiontype.ClientConnectionType
 	SendPacketToClient(packet d2netpacket.NetPacket) error
 	GetPlayerState() *d2player.PlayerState
 	SetPlayerState(playerState *d2player.PlayerState)

d2networking/d2server/connection_manager.go

@@ -0,0 +1,91 @@
+package d2server
+
+import (
+	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2client/d2clientconnectiontype"
+	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2netpacket"
+	"log"
+	"sync"
+	"time"
+)
+
+// ConnectionManager is responsible for cleanup up connections accepted by the game server. As the server communicates over
+// UDP and is stateless we need to implement some loose state management via a ping/pong system. ConnectionManager also handles
+// communication for graceful shutdowns.
+//
+// retries: # of attempts before the dropping the client
+// interval: How long to wait before each ping/pong test
+// gameServer: The *GameServer is argument provided for the connection manager to watch over
+// status: map of inflight ping/pong requests
+type ConnectionManager struct {
+	sync.RWMutex
+	retries    int
+	interval   time.Duration
+	gameServer *GameServer
+	status     map[string]int
+}
+
+func CreateConnectionManager(gameServer *GameServer) *ConnectionManager {
+	manager := &ConnectionManager{
+		retries:    3,
+		interval:   time.Millisecond * 1000,
+		gameServer: gameServer,
+		status:     make(map[string]int),
+	}
+
+	go manager.Run()
+
+	return manager
+}
+
+// Run starts up any watchers for for the connection manager
+func (c *ConnectionManager) Run() {
+	log.Print("Starting connection manager...")
+	for {
+		c.checkPeers()
+		time.Sleep(c.interval)
+	}
+}
+
+// checkPeers manages connection validation and cleanup for all peers.
+func (c *ConnectionManager) checkPeers() {
+	for id, connection := range c.gameServer.clientConnections {
+		if connection.GetConnectionType() != d2clientconnectiontype.Local {
+			if err := connection.SendPacketToClient(d2netpacket.CreatePingPacket()); err != nil {
+				log.Printf("Cannot ping client id: %s", id)
+			}
+			c.RWMutex.Lock()
+			c.status[id] += 1
+
+			if c.status[id] >= c.retries {
+				delete(c.status, id)
+				c.Drop(id)
+			}
+
+			c.RWMutex.Unlock()
+		}
+	}
+}
+
+// Recv simply resets the counter, acknowledging we have received a pong from the client.
+func (c *ConnectionManager) Recv(id string) {
+	c.status[id] = 0
+}
+
+// Drop removes the client id from the connection pool of the game server.
+func (c *ConnectionManager) Drop(id string) {
+	c.gameServer.RWMutex.Lock()
+	defer c.gameServer.RWMutex.Unlock()
+	delete(c.gameServer.clientConnections, id)
+	log.Printf("%s has been disconnected...", id)
+}
+
+// Shutdown will notify all of the clients that the server has been shutdown.
+func (c *ConnectionManager) Shutdown() {
+	// TODO: Currently this will never actually get called as the go routines are never signaled about the application termination.
+	// Things can be done more cleanly once we have graceful exits however we still need to account for other OS Signals
+	log.Print("Notifying clients server is shutting down...")
+	for _, connection := range c.gameServer.clientConnections {
+		connection.SendPacketToClient(d2netpacket.CreateServerClosedPacket())
+	}
+	Stop()
+}

d2networking/d2server/d2udpclientconnection/udp_client_connection.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"compress/gzip"
 	"encoding/json"
+	"github.com/OpenDiablo2/OpenDiablo2/d2networking/d2client/d2clientconnectiontype"
 	"net"
 
 	"github.com/OpenDiablo2/OpenDiablo2/d2game/d2player"
@@ -32,8 +33,8 @@ func (u UDPClientConnection) GetUniqueId() string {
 	return u.id
 }
 
-func (u UDPClientConnection) GetConnectionType() string {
-	return "Remote Client"
+func (u UDPClientConnection) GetConnectionType() d2clientconnectiontype.ClientConnectionType {
+	return d2clientconnectiontype.LANClient
 }
 
 func (u *UDPClientConnection) SendPacketToClient(packet d2netpacket.NetPacket) error {
@@ -46,7 +47,10 @@ func (u *UDPClientConnection) SendPacketToClient(packet d2netpacket.NetPacket) e
 	writer, _ := gzip.NewWriterLevel(&buff, gzip.BestCompression)
 	writer.Write(data)
 	writer.Close()
-	u.udpConnection.WriteToUDP(buff.Bytes(), u.address)
+	_, err = u.udpConnection.WriteToUDP(buff.Bytes(), u.address)
+	if err != nil {
+		return err
+	}
 
 	return nil
 }

d2networking/d2server/game_server.go

@@ -9,6 +9,7 @@ import (
 	"log"
 	"net"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/OpenDiablo2/OpenDiablo2/d2core/d2map/d2mapgen"
@@ -24,7 +25,9 @@ import (
 )
 
 type GameServer struct {
+	sync.RWMutex
 	clientConnections map[string]ClientConnection
+	manager           *ConnectionManager
 	mapEngines        []*d2mapengine.MapEngine
 	scriptEngine      *d2script.ScriptEngine
 	udpConnection     *net.UDPConn
@@ -47,6 +50,8 @@ func Create(openNetworkServer bool) {
 		seed:              time.Now().UnixNano(),
 	}
 
+	singletonServer.manager = CreateConnectionManager(singletonServer)
+
 	mapEngine := d2mapengine.CreateMapEngine()
 	mapEngine.SetSeed(singletonServer.seed)
 	mapEngine.ResetMap(d2enum.RegionAct1Town, 100, 100) // TODO: Mapgen - Needs levels.txt stuff
@@ -112,8 +117,17 @@ func runNetworkServer() {
 			for _, player := range singletonServer.clientConnections {
 				player.SendPacketToClient(netPacket)
 			}
+		case d2netpackettype.Pong:
+			packetData := d2netpacket.PlayerConnectionRequestPacket{}
+			json.Unmarshal([]byte(stringData), &packetData)
+			singletonServer.manager.Recv(packetData.Id)
+		case d2netpackettype.ServerClosed:
+			singletonServer.manager.Shutdown()
+		case d2netpackettype.PlayerDisconnectionNotification:
+			var packet d2netpacket.PlayerDisconnectRequestPacket
+			json.Unmarshal([]byte(stringData), &packet)
+			log.Printf("Received disconnect: %s", packet.Id)
 		}
-
 	}
 }