mirror of
https://github.com/psforever/PSF-LoginServer.git
synced 2026-01-19 18:44:45 +00:00
Multiple Game World Ports (#1206)
* configuration for the game world server to connect to clients across a number of socket-port connections following a simple numerical load balancing policy * combining port management into the sector pane * mostly spelling issues
This commit is contained in:
Fate-JH
GitHub
parent
593caec8cf
commit
00a6f2abe1
|
|
@ -14,7 +14,7 @@ import akka.{actor => classic}
|
|||
import ch.qos.logback.classic.LoggerContext
|
||||
import ch.qos.logback.classic.joran.JoranConfigurator
|
||||
import io.sentry.{Sentry, SentryOptions}
|
||||
import net.psforever.actors.net.{LoginActor, MiddlewareActor, SocketActor}
|
||||
import net.psforever.actors.net.{LoginActor, MiddlewareActor, SocketSetup, SocketSetupInfo, SocketPane}
|
||||
import net.psforever.actors.session.SessionActor
|
||||
import net.psforever.login.psadmin.PsAdminActor
|
||||
import net.psforever.login._
|
||||
|
|
@ -98,31 +98,7 @@ object Server {
|
|||
implicit val system: ActorSystem = classic.ActorSystem("PsLogin")
|
||||
Default(system)
|
||||
|
||||
// typed to classic wrappers for login and session actors
|
||||
val login = (ref: ActorRef[MiddlewareActor.Command], info: InetSocketAddress, connectionId: String) => {
|
||||
import net.psforever.services.account.IPAddress
|
||||
Behaviors.setup[PlanetSidePacket](context => {
|
||||
val actor = context.actorOf(classic.Props(new LoginActor(ref, connectionId, Login.getNewId())), "login")
|
||||
actor ! ReceiveIPAddress(new IPAddress(info))
|
||||
Behaviors.receiveMessage(message => {
|
||||
actor ! message
|
||||
Behaviors.same
|
||||
})
|
||||
})
|
||||
}
|
||||
val session = (ref: ActorRef[MiddlewareActor.Command], info: InetSocketAddress, connectionId: String) => {
|
||||
Behaviors.setup[PlanetSidePacket](context => {
|
||||
val uuid = randomUUID().toString
|
||||
val actor =
|
||||
context.actorOf(classic.Props(new SessionActor(ref, connectionId, Session.getNewId())), s"session-$uuid")
|
||||
Behaviors.receiveMessage(message => {
|
||||
actor ! message
|
||||
Behaviors.same
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
val zones = Zones.zones ++ Seq(Zone.Nowhere)
|
||||
val zones = Zones.zones :+ Zone.Nowhere
|
||||
val serviceManager = ServiceManager.boot
|
||||
serviceManager ! ServiceManager.Register(classic.Props[AccountIntermediaryService](), "accountIntermediary")
|
||||
serviceManager ! ServiceManager.Register(classic.Props[GalaxyService](), "galaxy")
|
||||
|
|
@ -135,8 +111,36 @@ object Server {
|
|||
system.spawn(InterstellarClusterService(zones), InterstellarClusterService.InterstellarClusterServiceKey.id)
|
||||
system.spawn(ChatService(), ChatService.ChatServiceKey.id)
|
||||
|
||||
system.spawn(SocketActor(new InetSocketAddress(bindAddress, Config.app.login.port), login), "login-socket")
|
||||
system.spawn(SocketActor(new InetSocketAddress(bindAddress, Config.app.world.port), session), "world-socket")
|
||||
// typed to classic wrappers for login and session actors
|
||||
val loginPlan = (ref: ActorRef[MiddlewareActor.Command], info: InetSocketAddress, connectionId: String) => {
|
||||
import net.psforever.services.account.IPAddress
|
||||
Behaviors.setup[PlanetSidePacket](context => {
|
||||
val actor = context.actorOf(classic.Props(new LoginActor(ref, connectionId, Login.getNewId())), "login")
|
||||
actor ! ReceiveIPAddress(new IPAddress(info))
|
||||
Behaviors.receiveMessage(message => {
|
||||
actor ! message
|
||||
Behaviors.same
|
||||
})
|
||||
})
|
||||
}
|
||||
val sessionPlan = (ref: ActorRef[MiddlewareActor.Command], info: InetSocketAddress, connectionId: String) => {
|
||||
Behaviors.setup[PlanetSidePacket](context => {
|
||||
val uuid = randomUUID().toString
|
||||
val actor =
|
||||
context.actorOf(classic.Props(new SessionActor(ref, connectionId, Session.getNewId())), s"session-$uuid")
|
||||
Behaviors.receiveMessage(message => {
|
||||
actor ! message
|
||||
Behaviors.same
|
||||
})
|
||||
})
|
||||
}
|
||||
system.spawn(
|
||||
SocketPane(Seq(
|
||||
SocketSetup("login", SocketSetupInfo(bindAddress, Seq(Config.app.login.port), loginPlan)),
|
||||
SocketSetup("world", SocketSetupInfo(bindAddress, Config.app.world.port +: Config.app.world.ports, sessionPlan))
|
||||
)),
|
||||
name = SocketPane.SocketPaneKey.id
|
||||
)
|
||||
|
||||
val adminListener = system.actorOf(
|
||||
classic.Props(
|
||||
|
|
|
|||
|
|
@ -20,6 +20,7 @@ login {
|
|||
world {
|
||||
# UDP listening port
|
||||
port = 51001
|
||||
ports = [51003, 51004, 51005, 51006, 51007, 51008, 51009, 51010]
|
||||
|
||||
# The name of the server as displayed in the server browser.
|
||||
server-name = PSForever
|
||||
|
|
|
|||
|
|
@ -1,7 +1,10 @@
|
|||
package net.psforever.actors.net
|
||||
|
||||
import akka.actor.typed.receptionist.Receptionist
|
||||
|
||||
import java.net.{InetAddress, InetSocketAddress}
|
||||
import akka.actor.{Actor, ActorRef, Cancellable, MDCContextAware, typed}
|
||||
import akka.actor.typed.scaladsl.adapter._
|
||||
import com.github.t3hnar.bcrypt._
|
||||
import net.psforever.objects.{Account, Default}
|
||||
import net.psforever.packet.PlanetSideGamePacket
|
||||
|
|
@ -17,34 +20,19 @@ import net.psforever.util.Database._
|
|||
|
||||
import java.security.MessageDigest
|
||||
import org.joda.time.LocalDateTime
|
||||
|
||||
import scala.collection.mutable
|
||||
import scala.concurrent.Future
|
||||
import scala.concurrent.duration._
|
||||
import scala.util.matching.Regex
|
||||
import scala.util.{Failure, Success}
|
||||
|
||||
/*
|
||||
object LoginActor {
|
||||
def apply(
|
||||
middlewareActor: typed.ActorRef[MiddlewareActor.Command],
|
||||
uuid: String
|
||||
): Behavior[Command] =
|
||||
Behaviors.setup(context => new LoginActor(context, middlewareActor, uuid).start())
|
||||
|
||||
sealed trait Command
|
||||
|
||||
final case class ReceptionistListing(listing: Receptionist.Listing) extends Command
|
||||
}
|
||||
|
||||
class LoginActor(
|
||||
middlewareActor: typed.ActorRef[MiddlewareActor.Command],
|
||||
uuid: String
|
||||
) {
|
||||
|
||||
def start(): Unit = {
|
||||
Behaviors.receiveMessagePartial {}
|
||||
}
|
||||
}
|
||||
|
||||
*/
|
||||
|
||||
class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], connectionId: String, sessionId: Long)
|
||||
extends Actor
|
||||
with MDCContextAware {
|
||||
|
|
@ -53,11 +41,12 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
|
||||
private case class UpdateServerList()
|
||||
|
||||
val usernameRegex = """[A-Za-z0-9]{3,}""".r
|
||||
val usernameRegex: Regex = """[A-Za-z0-9]{3,}""".r
|
||||
|
||||
var leftRef: ActorRef = ActorRef.noSender
|
||||
var rightRef: ActorRef = ActorRef.noSender
|
||||
var accountIntermediary: ActorRef = ActorRef.noSender
|
||||
var leftRef: ActorRef = Default.Actor
|
||||
var rightRef: ActorRef = Default.Actor
|
||||
var accountIntermediary: ActorRef = Default.Actor
|
||||
var sockets: typed.ActorRef[SocketPane.Command] = Default.typed.Actor
|
||||
|
||||
var updateServerListTask: Cancellable = Default.Cancellable
|
||||
|
||||
|
|
@ -66,14 +55,15 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
var canonicalHostName: String = ""
|
||||
var port: Int = 0
|
||||
|
||||
val serverName = Config.app.world.serverName
|
||||
val serverName: String = Config.app.world.serverName
|
||||
val publicAddress = new InetSocketAddress(InetAddress.getByName(Config.app.public), Config.app.world.port)
|
||||
|
||||
private val bcryptRounds = 12
|
||||
|
||||
ServiceManager.serviceManager ! Lookup("accountIntermediary")
|
||||
ServiceManager.receptionist ! Receptionist.Find(SocketPane.SocketPaneKey, context.self)
|
||||
|
||||
override def postStop() = {
|
||||
override def postStop(): Unit = {
|
||||
if (updateServerListTask != null)
|
||||
updateServerListTask.cancel()
|
||||
}
|
||||
|
|
@ -81,38 +71,46 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
def receive: Receive = {
|
||||
case ServiceManager.LookupResult("accountIntermediary", endpoint) =>
|
||||
accountIntermediary = endpoint
|
||||
|
||||
case SocketPane.SocketPaneKey.Listing(listings) =>
|
||||
sockets = listings.head
|
||||
|
||||
case ReceiveIPAddress(address) =>
|
||||
ipAddress = address.Address
|
||||
hostName = address.HostName
|
||||
canonicalHostName = address.CanonicalHostName
|
||||
port = address.Port
|
||||
|
||||
case UpdateServerList() =>
|
||||
updateServerList()
|
||||
|
||||
case packet: PlanetSideGamePacket =>
|
||||
handleGamePkt(packet)
|
||||
case default => failWithError(s"Invalid packet class received: $default")
|
||||
|
||||
case SocketPane.NextPort(_, address, portNum) =>
|
||||
val response = ConnectToWorldMessage(serverName, address.getHostAddress, portNum)
|
||||
middlewareActor ! MiddlewareActor.Send(response)
|
||||
middlewareActor ! MiddlewareActor.Close()
|
||||
|
||||
case default =>
|
||||
failWithError(s"Invalid packet class received: $default")
|
||||
}
|
||||
|
||||
def handleGamePkt(pkt: PlanetSideGamePacket) =
|
||||
def handleGamePkt(pkt: PlanetSideGamePacket): Unit =
|
||||
pkt match {
|
||||
case LoginMessage(majorVersion, minorVersion, buildDate, username, password, token, revision) =>
|
||||
// TODO: prevent multiple LoginMessages from being processed in a row!! We need a state machine
|
||||
|
||||
val clientVersion = s"Client Version: $majorVersion.$minorVersion.$revision, $buildDate"
|
||||
|
||||
if (token.isDefined)
|
||||
log.debug(s"New login UN:$username Token:${token.get}. $clientVersion")
|
||||
else {
|
||||
log.debug(s"New login UN:$username. $clientVersion")
|
||||
}
|
||||
|
||||
getAccountLogin(username, password, token)
|
||||
requestAccountLogin(username, password, token)
|
||||
|
||||
case ConnectToWorldRequestMessage(name, _, _, _, _, _, _, _) =>
|
||||
log.info(s"Connect to world request for '$name'")
|
||||
val response = ConnectToWorldMessage(serverName, publicAddress.getAddress.getHostAddress, publicAddress.getPort)
|
||||
middlewareActor ! MiddlewareActor.Send(response)
|
||||
middlewareActor ! MiddlewareActor.Close()
|
||||
sockets ! SocketPane.GetNextPort("world", context.self)
|
||||
|
||||
case _ =>
|
||||
log.warning(s"Unhandled GamePacket $pkt")
|
||||
|
|
@ -121,23 +119,19 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
// generates a password from username and password combination
|
||||
// mimics the process the launcher follows and hashes the password salted by the username
|
||||
def generateNewPassword(username: String, password: String): String = {
|
||||
|
||||
// salt password hash with username (like the launcher does) (username + password)
|
||||
val saltedPassword = username.concat(password)
|
||||
|
||||
// https://stackoverflow.com/a/46332228
|
||||
// hash password (like the launcher sends)
|
||||
val hashedPassword = MessageDigest.getInstance("SHA-256")
|
||||
.digest(saltedPassword.getBytes("UTF-8"))
|
||||
.map("%02x".format(_)).mkString
|
||||
|
||||
// bcrypt hash for DB storage
|
||||
val bcryptedPassword = hashedPassword.bcryptBounded(bcryptRounds)
|
||||
|
||||
bcryptedPassword
|
||||
}
|
||||
|
||||
def getAccountLogin(username: String, passwordOpt: Option[String], tokenOpt: Option[String]): Unit = {
|
||||
def requestAccountLogin(username: String, passwordOpt: Option[String], tokenOpt: Option[String]): Unit = {
|
||||
tokenOpt match {
|
||||
case Some(token) => accountLoginWithToken(token)
|
||||
case None => accountLogin(username, passwordOpt.getOrElse(""))
|
||||
|
|
@ -145,9 +139,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
}
|
||||
|
||||
def accountLogin(username: String, password: String): Unit = {
|
||||
|
||||
import ctx._
|
||||
|
||||
val newToken = this.generateToken()
|
||||
val result = for {
|
||||
// backwards compatibility: prefer exact match first, then try lowercase
|
||||
|
|
@ -158,21 +150,16 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
case Some(_) =>
|
||||
Future.successful(Seq())
|
||||
}
|
||||
|
||||
accountOption <- accountsExact.headOption orElse accountsLower.headOption match {
|
||||
|
||||
// account found
|
||||
case Some(account) =>
|
||||
Future.successful(Some(account))
|
||||
|
||||
// create new account
|
||||
case None =>
|
||||
if (Config.app.login.createMissingAccounts) {
|
||||
|
||||
// generate bcrypted passwords
|
||||
val bcryptedPassword = generateNewPassword(username, password)
|
||||
val passhash = password.bcryptBounded(bcryptRounds)
|
||||
|
||||
// save bcrypted password hash to DB
|
||||
ctx.run(
|
||||
query[persistence.Account]
|
||||
|
|
@ -185,7 +172,6 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
) flatMap { id => ctx.run(query[persistence.Account].filter(_.id == lift(id))) } map { accounts =>
|
||||
Some(accounts.head)
|
||||
}
|
||||
|
||||
} else {
|
||||
loginFailureResponse(username, newToken)
|
||||
Future.successful(None)
|
||||
|
|
@ -194,14 +180,11 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
|
||||
login <- accountOption match {
|
||||
case Some(account) =>
|
||||
|
||||
// remember: this is the in client "StagingTest" login handling
|
||||
// the password is send in clear and needs to be checked against the "old" (only bcrypted) passhash
|
||||
// if there ever is a way to update the password in the future passhash and password need be updated
|
||||
(account.inactive, password.isBcryptedBounded(account.passhash)) match {
|
||||
|
||||
case (false, true) =>
|
||||
|
||||
accountIntermediary ! StoreAccountData(newToken, Account(account.id, account.username, account.gm))
|
||||
val future = ctx.run(
|
||||
query[persistence.Login].insert(
|
||||
|
|
@ -212,14 +195,11 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
_.port -> lift(port)
|
||||
)
|
||||
)
|
||||
|
||||
// handle new password
|
||||
if (account.password == "") {
|
||||
|
||||
// generate bcrypted password
|
||||
// use username as provided by the user (db entry could be wrong), that is the way the launcher does it
|
||||
val bcryptedPassword = generateNewPassword(username, password)
|
||||
|
||||
// update account, set password
|
||||
ctx.run(
|
||||
query[persistence.Account]
|
||||
|
|
@ -227,21 +207,21 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
.update(_.password -> lift(bcryptedPassword))
|
||||
)
|
||||
}
|
||||
|
||||
loginSuccessfulResponse(username, newToken)
|
||||
updateServerListTask =
|
||||
context.system.scheduler.scheduleWithFixedDelay(0 seconds, 5 seconds, self, UpdateServerList())
|
||||
future
|
||||
|
||||
case (_, false) =>
|
||||
loginPwdFailureResponse(username, newToken)
|
||||
Future.successful(None)
|
||||
|
||||
case (true, _) =>
|
||||
loginAccountFailureResponse(username, newToken)
|
||||
Future.successful(None)
|
||||
}
|
||||
case None => Future.successful(None)
|
||||
}
|
||||
|
||||
} yield login
|
||||
|
||||
result.onComplete {
|
||||
|
|
@ -251,18 +231,12 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
}
|
||||
|
||||
def accountLoginWithToken(token: String): Unit = {
|
||||
|
||||
import ctx._
|
||||
|
||||
val newToken = this.generateToken()
|
||||
val result = for {
|
||||
|
||||
accountsExact <- ctx.run(query[persistence.Account].filter(_.token.getOrNull == lift(token)))
|
||||
|
||||
accountOption <- accountsExact.headOption match {
|
||||
|
||||
case Some(account) =>
|
||||
|
||||
// token expires after 2 hours
|
||||
// new connections and players leaving a world server will return to desktop
|
||||
if (LocalDateTime.now().isAfter(account.tokenCreated.get.plusHours(2))) {
|
||||
|
|
@ -280,9 +254,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
login <- accountOption match {
|
||||
case Some(account) =>
|
||||
(account.inactive, account.token.getOrElse("") == token) match {
|
||||
|
||||
case (false, true) =>
|
||||
|
||||
accountIntermediary ! StoreAccountData(newToken, Account(account.id, account.username, account.gm))
|
||||
val future = ctx.run(
|
||||
query[persistence.Login].insert(
|
||||
|
|
@ -293,7 +265,6 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
_.port -> lift(port)
|
||||
)
|
||||
)
|
||||
|
||||
loginSuccessfulResponseToken(account.username, token, newToken)
|
||||
updateServerListTask =
|
||||
context.system.scheduler.scheduleWithFixedDelay(0 seconds, 5 seconds, self, UpdateServerList())
|
||||
|
|
@ -307,10 +278,8 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
loginAccountFailureResponseToken(account.username, token, newToken)
|
||||
Future.successful(None)
|
||||
}
|
||||
|
||||
case None => Future.successful(None)
|
||||
}
|
||||
|
||||
} yield login
|
||||
|
||||
result.onComplete {
|
||||
|
|
@ -319,7 +288,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
}
|
||||
}
|
||||
|
||||
def loginSuccessfulResponse(username: String, newToken: String) = {
|
||||
def loginSuccessfulResponse(username: String, newToken: String): Unit = {
|
||||
middlewareActor ! MiddlewareActor.Send(
|
||||
LoginRespMessage(
|
||||
newToken,
|
||||
|
|
@ -333,10 +302,8 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
)
|
||||
}
|
||||
|
||||
def loginSuccessfulResponseToken(username: String, token: String, newToken: String) = {
|
||||
|
||||
def loginSuccessfulResponseToken(username: String, token: String, newToken: String): Unit = {
|
||||
log.info(s"User $username logged in unsing token $token")
|
||||
|
||||
middlewareActor ! MiddlewareActor.Send(
|
||||
LoginRespMessage(
|
||||
newToken,
|
||||
|
|
@ -350,7 +317,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
)
|
||||
}
|
||||
|
||||
def loginPwdFailureResponse(username: String, newToken: String) = {
|
||||
def loginPwdFailureResponse(username: String, newToken: String): Unit = {
|
||||
log.warning(s"Failed login to account $username")
|
||||
middlewareActor ! MiddlewareActor.Send(
|
||||
LoginRespMessage(
|
||||
|
|
@ -365,7 +332,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
)
|
||||
}
|
||||
|
||||
def loginFailureResponseToken(token: String, newToken: String) = {
|
||||
def loginFailureResponseToken(token: String, newToken: String): Unit = {
|
||||
log.warning(s"Failed login using unknown token $token")
|
||||
middlewareActor ! MiddlewareActor.Send(
|
||||
LoginRespMessage(
|
||||
|
|
@ -380,7 +347,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
)
|
||||
}
|
||||
|
||||
def loginFailureResponseTokenExpired(token: String, newToken: String) = {
|
||||
def loginFailureResponseTokenExpired(token: String, newToken: String): Unit = {
|
||||
log.warning(s"Failed login using expired token $token")
|
||||
middlewareActor ! MiddlewareActor.Send(
|
||||
LoginRespMessage(
|
||||
|
|
@ -395,7 +362,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
)
|
||||
}
|
||||
|
||||
def loginFailureResponse(username: String, newToken: String) = {
|
||||
def loginFailureResponse(username: String, newToken: String): Unit = {
|
||||
log.warning(s"DB problem username: $username")
|
||||
middlewareActor ! MiddlewareActor.Send(
|
||||
LoginRespMessage(
|
||||
|
|
@ -410,7 +377,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
)
|
||||
}
|
||||
|
||||
def loginFailureResponseToken(username: String, token: String, newToken: String) = {
|
||||
def loginFailureResponseToken(username: String, token: String, newToken: String): Unit = {
|
||||
log.warning(s"DB problem username $username token: $token")
|
||||
middlewareActor ! MiddlewareActor.Send(
|
||||
LoginRespMessage(
|
||||
|
|
@ -425,7 +392,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
)
|
||||
}
|
||||
|
||||
def loginAccountFailureResponse(username: String, newToken: String) = {
|
||||
def loginAccountFailureResponse(username: String, newToken: String): Unit = {
|
||||
log.warning(s"Account $username inactive")
|
||||
middlewareActor ! MiddlewareActor.Send(
|
||||
LoginRespMessage(
|
||||
|
|
@ -440,7 +407,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
)
|
||||
}
|
||||
|
||||
def loginAccountFailureResponseToken(username: String, token: String, newToken: String) = {
|
||||
def loginAccountFailureResponseToken(username: String, token: String, newToken: String): Unit = {
|
||||
log.warning(s"Account $username inactive token: $token ")
|
||||
middlewareActor ! MiddlewareActor.Send(
|
||||
LoginRespMessage(
|
||||
|
|
@ -455,16 +422,16 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
)
|
||||
}
|
||||
|
||||
def generateToken() = {
|
||||
def generateToken(): String = {
|
||||
val r = new scala.util.Random
|
||||
val sb = new StringBuilder
|
||||
val sb = new mutable.StringBuilder
|
||||
for (_ <- 1 to 31) {
|
||||
sb.append(r.nextPrintableChar())
|
||||
}
|
||||
sb.toString
|
||||
}
|
||||
|
||||
def updateServerList() = {
|
||||
def updateServerList(): Unit = {
|
||||
middlewareActor ! MiddlewareActor.Send(
|
||||
VNLWorldStatusMessage(
|
||||
"Welcome to PlanetSide! ",
|
||||
|
|
@ -473,7 +440,7 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
serverName,
|
||||
WorldStatus.Up,
|
||||
Config.app.world.serverType,
|
||||
Vector(WorldConnectionInfo(publicAddress)),
|
||||
Vector(WorldConnectionInfo(publicAddress)), //todo ideally, ask for info from SocketPane
|
||||
PlanetSideEmpire.VS
|
||||
)
|
||||
)
|
||||
|
|
@ -485,5 +452,4 @@ class LoginActor(middlewareActor: typed.ActorRef[MiddlewareActor.Command], conne
|
|||
log.error(error)
|
||||
middlewareActor ! MiddlewareActor.Close()
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,58 @@
|
|||
// Copyright (c) 2024 PSForever
|
||||
package net.psforever.actors.net
|
||||
|
||||
import akka.actor.typed.{ActorRef, Behavior}
|
||||
import akka.actor.typed.scaladsl.{AbstractBehavior, ActorContext, Behaviors}
|
||||
import akka.io.Udp
|
||||
import net.psforever.util.Config
|
||||
|
||||
import java.util.concurrent.ThreadLocalRandom
|
||||
import scala.util.Random
|
||||
import scala.concurrent.duration._
|
||||
|
||||
// TODO? This doesn't quite support all parameters of the old network simulator
|
||||
// Need to decide wheter they are necessary or not
|
||||
// https://github.com/psforever/PSF-LoginServer/blob/07f447c2344ab55d581317316c41571772ac2242/src/main/scala/net/psforever/login/UdpNetworkSimulator.scala
|
||||
private object NetworkSimulator {
|
||||
def apply(socketActor: ActorRef[SocketActor.Command]): Behavior[Udp.Message] =
|
||||
Behaviors.setup(context => new NetworkSimulator(context, socketActor))
|
||||
}
|
||||
|
||||
private class NetworkSimulator(context: ActorContext[Udp.Message], socketActor: ActorRef[SocketActor.Command])
|
||||
extends AbstractBehavior[Udp.Message](context) {
|
||||
|
||||
private[this] val log = org.log4s.getLogger
|
||||
|
||||
override def onMessage(message: Udp.Message): Behavior[Udp.Message] = {
|
||||
message match {
|
||||
case _: Udp.Received | _: Udp.Send =>
|
||||
simulate(message)
|
||||
Behaviors.same
|
||||
case _ =>
|
||||
socketActor ! toSocket(message)
|
||||
Behaviors.same
|
||||
}
|
||||
}
|
||||
|
||||
def simulate(message: Udp.Message): Unit = {
|
||||
if (Random.nextDouble() > Config.app.development.netSim.loss) {
|
||||
if (Random.nextDouble() <= Config.app.development.netSim.reorderChance) {
|
||||
context.scheduleOnce(
|
||||
ThreadLocalRandom.current().nextDouble(0.01, 0.2).seconds,
|
||||
socketActor,
|
||||
toSocket(message)
|
||||
)
|
||||
} else {
|
||||
socketActor ! toSocket(message)
|
||||
}
|
||||
} else {
|
||||
log.trace("Network simulator dropped packet")
|
||||
}
|
||||
}
|
||||
|
||||
def toSocket(message: Udp.Message): SocketActor.Command =
|
||||
message match {
|
||||
case message: Udp.Command => SocketActor.UdpCommandMessage(message)
|
||||
case message: Udp.Event => SocketActor.UdpEventMessage(message)
|
||||
}
|
||||
}
|
||||
|
|
@ -3,12 +3,10 @@ package net.psforever.actors.net
|
|||
import java.net.InetSocketAddress
|
||||
import java.security.SecureRandom
|
||||
import java.util.UUID.randomUUID
|
||||
import java.util.concurrent.ThreadLocalRandom
|
||||
|
||||
import akka.actor.Cancellable
|
||||
import akka.{actor => classic}
|
||||
import akka.actor.typed.{ActorRef, ActorTags, Behavior, PostStop, Terminated}
|
||||
import akka.actor.typed.scaladsl.{AbstractBehavior, ActorContext, Behaviors}
|
||||
import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
|
||||
import akka.io.{IO, Udp}
|
||||
import akka.actor.typed.scaladsl.adapter._
|
||||
import net.psforever.packet.PlanetSidePacket
|
||||
|
|
@ -17,26 +15,32 @@ import scodec.interop.akka.EnrichedByteString
|
|||
|
||||
import scala.collection.mutable
|
||||
import scala.concurrent.ExecutionContextExecutor
|
||||
import scala.concurrent.duration.{DurationDouble, DurationInt}
|
||||
import scala.util.Random
|
||||
import scala.concurrent.duration.DurationInt
|
||||
|
||||
/** SocketActor creates a UDP socket, receives packets and forwards them to MiddlewareActor
|
||||
* There is only one SocketActor, but each connected client gets its own MiddlewareActor
|
||||
*/
|
||||
/**
|
||||
* Create a networking port attachment that accepts user datagram protocol (UDP) packets
|
||||
* and forwards those packets to a business logic unit referred to herein as a "plan".
|
||||
* The landing site of a plan is a processing entity of middleware logic that dissects composed packet data
|
||||
* and pushes that further down the chain to the business logic.
|
||||
* Each instance of middleware support and then business logic
|
||||
* is associated with a unique clients that attempt to connect to the server though this socket port.
|
||||
*/
|
||||
object SocketActor {
|
||||
def apply(
|
||||
address: InetSocketAddress,
|
||||
next: (ActorRef[MiddlewareActor.Command], InetSocketAddress, String) => Behavior[PlanetSidePacket]
|
||||
): Behavior[Command] =
|
||||
Behaviors.setup(context => new SocketActor(context, address, next).start())
|
||||
address: InetSocketAddress,
|
||||
nextPlan: (ActorRef[MiddlewareActor.Command], InetSocketAddress, String) => Behavior[PlanetSidePacket]
|
||||
): Behavior[Command] =
|
||||
Behaviors.setup(context => new SocketActor(context, address, nextPlan).start())
|
||||
|
||||
sealed trait Command
|
||||
|
||||
private final case class UdpCommandMessage(message: Udp.Command) extends Command
|
||||
private final case class UdpEventMessage(message: Udp.Event) extends Command
|
||||
private[net] final case class UdpCommandMessage(message: Udp.Command) extends Command
|
||||
private[net] final case class UdpEventMessage(message: Udp.Event) extends Command
|
||||
private final case class UdpUnboundMessage(message: Udp.Unbound) extends Command
|
||||
private final case class Bound(socket: classic.ActorRef) extends Command
|
||||
private final case class StopChild(ref: ActorRef[MiddlewareActor.Command]) extends Command
|
||||
final case class AskSocketLoad(replyTo: ActorRef[Any]) extends Command
|
||||
final case class SocketLoad(sessions: Int)
|
||||
|
||||
// Typed actors cannot access sender but you can only get the socket that way
|
||||
private class SenderHack(ref: ActorRef[SocketActor.Command]) extends classic.Actor {
|
||||
|
|
@ -48,60 +52,13 @@ object SocketActor {
|
|||
context.system.terminate()
|
||||
}
|
||||
}
|
||||
|
||||
// TODO? This doesn't quite support all parameters of the old network simulator
|
||||
// Need to decide wheter they are necessary or not
|
||||
// https://github.com/psforever/PSF-LoginServer/blob/07f447c2344ab55d581317316c41571772ac2242/src/main/scala/net/psforever/login/UdpNetworkSimulator.scala
|
||||
private object NetworkSimulator {
|
||||
def apply(socketActor: ActorRef[SocketActor.Command]): Behavior[Udp.Message] =
|
||||
Behaviors.setup(context => new NetworkSimulator(context, socketActor))
|
||||
}
|
||||
|
||||
private class NetworkSimulator(context: ActorContext[Udp.Message], socketActor: ActorRef[SocketActor.Command])
|
||||
extends AbstractBehavior[Udp.Message](context) {
|
||||
|
||||
private[this] val log = org.log4s.getLogger
|
||||
|
||||
override def onMessage(message: Udp.Message): Behavior[Udp.Message] = {
|
||||
message match {
|
||||
case _: Udp.Received | _: Udp.Send =>
|
||||
simulate(message)
|
||||
Behaviors.same
|
||||
case _ =>
|
||||
socketActor ! toSocket(message)
|
||||
Behaviors.same
|
||||
}
|
||||
}
|
||||
|
||||
def simulate(message: Udp.Message): Unit = {
|
||||
if (Random.nextDouble() > Config.app.development.netSim.loss) {
|
||||
if (Random.nextDouble() <= Config.app.development.netSim.reorderChance) {
|
||||
context.scheduleOnce(
|
||||
ThreadLocalRandom.current().nextDouble(0.01, 0.2).seconds,
|
||||
socketActor,
|
||||
toSocket(message)
|
||||
)
|
||||
} else {
|
||||
socketActor ! toSocket(message)
|
||||
}
|
||||
} else {
|
||||
log.trace("Network simulator dropped packet")
|
||||
}
|
||||
}
|
||||
|
||||
def toSocket(message: Udp.Message): Command =
|
||||
message match {
|
||||
case message: Udp.Command => UdpCommandMessage(message)
|
||||
case message: Udp.Event => UdpEventMessage(message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class SocketActor(
|
||||
context: ActorContext[SocketActor.Command],
|
||||
address: InetSocketAddress,
|
||||
next: (ActorRef[MiddlewareActor.Command], InetSocketAddress, String) => Behavior[PlanetSidePacket]
|
||||
) {
|
||||
context: ActorContext[SocketActor.Command],
|
||||
address: InetSocketAddress,
|
||||
nextPlan: (ActorRef[MiddlewareActor.Command], InetSocketAddress, String) => Behavior[PlanetSidePacket]
|
||||
) {
|
||||
import SocketActor._
|
||||
import SocketActor.Command
|
||||
|
||||
|
|
@ -123,8 +80,8 @@ class SocketActor(
|
|||
context.spawnAnonymous(NetworkSimulator(context.self))
|
||||
}
|
||||
|
||||
val updUnboundAdapter: ActorRef[Udp.Unbound] = context.messageAdapter[Udp.Unbound](UdpUnboundMessage)
|
||||
val senderHack: classic.ActorRef = context.actorOf(classic.Props(new SenderHack(context.self)))
|
||||
//val updUnboundAdapter: ActorRef[Udp.Unbound] = context.messageAdapter[Udp.Unbound](UdpUnboundMessage)
|
||||
val senderHack: classic.ActorRef = context.actorOf(classic.Props(new SenderHack(context.self)))
|
||||
|
||||
IO(Udp)(context.system.classicSystem).tell(Udp.Bind(updEventAdapter.toClassic, address), senderHack)
|
||||
|
||||
|
|
@ -183,7 +140,7 @@ class SocketActor(
|
|||
case None =>
|
||||
val connectionId = randomUUID.toString
|
||||
val ref = context.spawn(
|
||||
MiddlewareActor(udpCommandAdapter, remote, next, connectionId),
|
||||
MiddlewareActor(udpCommandAdapter, remote, nextPlan, connectionId),
|
||||
s"middleware-$connectionId",
|
||||
ActorTags(s"uuid=$connectionId")
|
||||
)
|
||||
|
|
@ -208,13 +165,16 @@ class SocketActor(
|
|||
socket ! message
|
||||
Behaviors.same
|
||||
|
||||
case AskSocketLoad(replyTo) =>
|
||||
replyTo ! SocketLoad(packetActors.size)
|
||||
Behaviors.same
|
||||
|
||||
case UdpUnboundMessage(_) =>
|
||||
Behaviors.stopped
|
||||
|
||||
case StopChild(ref) =>
|
||||
context.stop(ref)
|
||||
Behaviors.same
|
||||
|
||||
}
|
||||
.receiveSignal {
|
||||
case (_, PostStop) =>
|
||||
|
|
|
|||
208
src/main/scala/net/psforever/actors/net/SocketPane.scala
Normal file
208
src/main/scala/net/psforever/actors/net/SocketPane.scala
Normal file
|
|
@ -0,0 +1,208 @@
|
|||
// Copyright (c) 2024 PSForever
|
||||
package net.psforever.actors.net
|
||||
|
||||
import akka.actor.typed.receptionist.{Receptionist, ServiceKey}
|
||||
|
||||
import java.net.{InetAddress, InetSocketAddress}
|
||||
import akka.{actor => classic}
|
||||
import akka.actor.typed.{ActorRef, Behavior, PostStop}
|
||||
import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
|
||||
import net.psforever.packet.PlanetSidePacket
|
||||
|
||||
private[net] object SocketPanePortRotation {
|
||||
/**
|
||||
* Overrode constructor for `SocketPanePortRotation` entities.
|
||||
* Copy constructor, essentially, that retains the internal current rotation index.
|
||||
* @param rotation the previous rotation entity
|
||||
* @return a copy of the previous rotation entity
|
||||
*/
|
||||
def apply(rotation: SocketPanePortRotation): SocketPanePortRotation = {
|
||||
SocketPanePortRotation(rotation.portNumbers, rotation.currentIndex)
|
||||
}
|
||||
|
||||
/**
|
||||
* Overrode constructor for `SocketPanePortRotation` entities.
|
||||
* Adda new port to the list of ports but retain the internal current rotation index.
|
||||
* @param rotation the previous rotation entity
|
||||
* @param newPort the new port number
|
||||
* @return a copy of the previous rotation entity with an additional port that can be selected
|
||||
*/
|
||||
def apply(rotation: SocketPanePortRotation, newPort: Int): SocketPanePortRotation = {
|
||||
SocketPanePortRotation(rotation.portNumbers :+ newPort, rotation.currentIndex)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* For a given sequence of ports,
|
||||
* cycle through port numbers until the last port number has been produced
|
||||
* then start from the first port number again.
|
||||
* @param portNumbers the sequence of ports to be cycled between
|
||||
* @param index optional;
|
||||
* the starting index in the sequence of ports;
|
||||
* default is 0
|
||||
*/
|
||||
private[net] case class SocketPanePortRotation(
|
||||
portNumbers: Seq[Int],
|
||||
index: Int = 0
|
||||
) {
|
||||
private var currentIndex: Int = index
|
||||
|
||||
/**
|
||||
* Retrieve the sequentially next port number.
|
||||
* @return the next port number
|
||||
*/
|
||||
def NextPort: Int = {
|
||||
val out = portNumbers.lift(currentIndex).orElse(portNumbers.headOption).getOrElse(0)
|
||||
currentIndex += 1
|
||||
currentIndex %= portNumbers.size
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Information explaining how to manage a socket group.
|
||||
* @param groupId moniker for the group
|
||||
* @param info how the sockets in this group operate
|
||||
* @param initial should this socket group be protected as "original";
|
||||
* defaults to `false`
|
||||
*/
|
||||
final case class SocketSetup(groupId: String, info: SocketSetupInfo, initial: Boolean = false) {
|
||||
assert(info.ports.nonEmpty, s"port group $groupId should define port numbers")
|
||||
/* the same port can belong to multiple groups, but the same port may not be repeated in the same group */
|
||||
assert(info.ports.size == info.ports.distinct.size, s"port group $groupId should not contain duplicate port numbers")
|
||||
}
|
||||
|
||||
/**
|
||||
* Information explaining the details of a socket group.
|
||||
* @param address Internet protocol location of the host
|
||||
* @param ports network ports that are used as endpoints of transmission;
|
||||
* corresponds to a number of sockets
|
||||
* @param planOfAction whenever a new connection across a socket is made, the method of consuming network packets
|
||||
*/
|
||||
final case class SocketSetupInfo(
|
||||
address: InetAddress,
|
||||
ports: Seq[Int],
|
||||
planOfAction: (ActorRef[MiddlewareActor.Command], InetSocketAddress, String) => Behavior[PlanetSidePacket]
|
||||
)
|
||||
|
||||
object SocketPane {
|
||||
val SocketPaneKey: ServiceKey[Command] = ServiceKey[SocketPane.Command](id = "socketPane")
|
||||
|
||||
/**
|
||||
* Overrode constructor for `SocketPane` entities.
|
||||
* Registers the entity with the actor receptionist.
|
||||
* @param setupInfo the details of the socket groups
|
||||
* @return a `SocketPane` entity
|
||||
*/
|
||||
def apply(setupInfo: Seq[SocketSetup]): Behavior[Command] = {
|
||||
Behaviors.setup { context =>
|
||||
context.system.receptionist ! Receptionist.Register(SocketPaneKey, context.self)
|
||||
new SocketPane(context, setupInfo).start()
|
||||
}
|
||||
}
|
||||
|
||||
sealed trait Command
|
||||
|
||||
final case class CreateNewSocket(groupId: String, port: Int) extends Command
|
||||
|
||||
final case class CreateNewSocketGroup(groupId: String, info: SocketSetupInfo) extends Command
|
||||
|
||||
final case class GetNextPort(groupId: String, replyTo: classic.ActorRef) extends Command
|
||||
|
||||
final case class NextPort(groupId: String, address: InetAddress, port: Int)
|
||||
}
|
||||
|
||||
/**
|
||||
* Management of sockets connecting to the network ports.
|
||||
* <br><br>
|
||||
* Connections to the networking ports are created by the logic imposed by this class
|
||||
* and are handled by sockets that bind to those ports and accept or pass packets to game logic.
|
||||
* This game logic is encapsulated by an anonymous function
|
||||
* that is automatically implemented into a game logic machine (consumption and production)
|
||||
* upon unique connections detected / attempted across those sockets.
|
||||
* Multiple sockets can connect to the same port so no compensation is required.
|
||||
* <br><br>
|
||||
* New sockets to ports can be added to existing groups after the initial socket groups.
|
||||
* New socket groups can be created after the initial information.
|
||||
* @param context hook for setting up the sockets and, eventually, their packet logic
|
||||
* @param initialPortSetup the details of the socket groups
|
||||
*/
|
||||
class SocketPane(
|
||||
context: ActorContext[SocketPane.Command],
|
||||
private val initialPortSetup: Seq[SocketSetup]
|
||||
) {
|
||||
private[this] val log = org.log4s.getLogger
|
||||
|
||||
/** original socket group information, now properly flagged as "original" */
|
||||
private var socketConfigs: Seq[SocketSetup] = initialPortSetup.map { setup => setup.copy(initial = true) }
|
||||
/** all sockets produced by the socket group information and any later socket creation commands */
|
||||
private var socketActors: Array[ActorRef[SocketActor.Command]] = initialPortSetup.flatMap {
|
||||
case SocketSetup(_, SocketSetupInfo(address, ports, plan), _) =>
|
||||
ports.map { portNum => context.spawn(SocketActor(new InetSocketAddress(address, portNum), plan), name=s"world-socket-$portNum") }
|
||||
}.toArray
|
||||
/** load balancing for redirecting newly discovered packet input to different sockets (ports);
|
||||
* should be referenced externally to switch sockets;
|
||||
* see SocketActor.GetNextPort */
|
||||
private var socketRotations: Array[SocketPanePortRotation] = initialPortSetup.map {
|
||||
case SocketSetup(_, SocketSetupInfo(_, ports, _), _) => SocketPanePortRotation(ports)
|
||||
}.toArray
|
||||
|
||||
log.debug(s"sockets configured for ${socketActors.length} ports initially")
|
||||
|
||||
def start(): Behavior[SocketPane.Command] = {
|
||||
Behaviors
|
||||
.receiveMessagePartial[SocketPane.Command] {
|
||||
case SocketPane.CreateNewSocket(key, _)
|
||||
if !socketConfigs.exists { setup => setup.groupId.equals(key) } =>
|
||||
log.warn(s"port group $key does not exist and can not be appended to")
|
||||
Behaviors.same
|
||||
|
||||
case SocketPane.CreateNewSocket(groupId, port)
|
||||
if socketConfigs
|
||||
.find { setup => setup.groupId.equals(groupId) }
|
||||
.exists { case SocketSetup(_, SocketSetupInfo(_, ports, _), _) => ports.contains(port) } =>
|
||||
log.info(s"new port $port for group $groupId already supported")
|
||||
Behaviors.same
|
||||
|
||||
case SocketPane.CreateNewSocket(groupId, port) =>
|
||||
log.info(s"new socket to port $port created in $groupId")
|
||||
val index = socketConfigs.indexWhere { setup => setup.groupId.equals(groupId) }
|
||||
val SocketSetup(_, SocketSetupInfo(address, ports, plan), _) = socketConfigs(index)
|
||||
socketActors = socketActors :+
|
||||
context.spawn(SocketActor(new InetSocketAddress(address, port), plan), name=s"world-socket-$port")
|
||||
socketConfigs = (socketConfigs.take(index) :+ SocketSetup(groupId, SocketSetupInfo(address, ports :+ port, plan))) ++
|
||||
socketConfigs.drop(index + 1)
|
||||
socketRotations = (socketRotations.take(index) :+ SocketPanePortRotation(socketRotations(index), port)) ++
|
||||
socketRotations.drop(index + 1)
|
||||
Behaviors.same
|
||||
|
||||
case SocketPane.CreateNewSocketGroup(groupId, _)
|
||||
if socketConfigs.exists { case SocketSetup(oldKey, _, _) => oldKey.equals(groupId) } =>
|
||||
log.warn(s"port group $groupId already exists and can not be created twice")
|
||||
Behaviors.same
|
||||
|
||||
case SocketPane.CreateNewSocketGroup(groupId, info @ SocketSetupInfo(address, ports, plan)) =>
|
||||
socketActors = socketActors ++
|
||||
ports.map { portNum => context.spawn(SocketActor(new InetSocketAddress(address, portNum), plan), name=s"world-socket-$portNum") }
|
||||
socketConfigs = socketConfigs :+
|
||||
SocketSetup(groupId, info)
|
||||
socketRotations = socketRotations :+
|
||||
SocketPanePortRotation(ports)
|
||||
Behaviors.same
|
||||
|
||||
case SocketPane.GetNextPort(groupId, replyTo) =>
|
||||
socketConfigs.indexWhere { setup => setup.groupId.equals(groupId) } match {
|
||||
case -1 =>
|
||||
log.warn(s"port group $groupId does not exist")
|
||||
case index =>
|
||||
replyTo ! SocketPane.NextPort(groupId, socketConfigs(index).info.address, socketRotations(index).NextPort)
|
||||
}
|
||||
Behaviors.same
|
||||
}
|
||||
.receiveSignal {
|
||||
case (_, PostStop) =>
|
||||
socketActors.foreach(context.stop)
|
||||
Behaviors.same
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -25,7 +25,7 @@ class AccountIntermediaryService extends Actor {
|
|||
private val IPAddressBySessionID = mutable.Map[Long, IPAddress]()
|
||||
private[this] val log = org.log4s.getLogger
|
||||
|
||||
def receive = {
|
||||
def receive: Receive = {
|
||||
// Called by the LoginSessionActor
|
||||
case StoreAccountData(token, account) =>
|
||||
accountsByToken += (token -> account)
|
||||
|
|
|
|||
|
|
@ -104,6 +104,7 @@ case class LoginConfig(
|
|||
|
||||
case class WorldConfig(
|
||||
port: Int,
|
||||
ports: Seq[Int],
|
||||
serverName: String,
|
||||
serverType: ServerType
|
||||
)
|
||||
|
|
|
|||
Loading…
Reference in a new issue