diff --git a/server/src/main/scala/net/psforever/server/Server.scala b/server/src/main/scala/net/psforever/server/Server.scala index f82469d7b..2ec30a503 100644 --- a/server/src/main/scala/net/psforever/server/Server.scala +++ b/server/src/main/scala/net/psforever/server/Server.scala @@ -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( diff --git a/src/main/resources/application.conf b/src/main/resources/application.conf index aa4dc3bda..dc15c07e2 100644 --- a/src/main/resources/application.conf +++ b/src/main/resources/application.conf @@ -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 diff --git a/src/main/scala/net/psforever/actors/net/LoginActor.scala b/src/main/scala/net/psforever/actors/net/LoginActor.scala index 3fe6cd46e..e359a2937 100644 --- a/src/main/scala/net/psforever/actors/net/LoginActor.scala +++ b/src/main/scala/net/psforever/actors/net/LoginActor.scala @@ -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() } - } diff --git a/src/main/scala/net/psforever/actors/net/NetworkSimulator.scala b/src/main/scala/net/psforever/actors/net/NetworkSimulator.scala new file mode 100644 index 000000000..3f2e35104 --- /dev/null +++ b/src/main/scala/net/psforever/actors/net/NetworkSimulator.scala @@ -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) + } +} diff --git a/src/main/scala/net/psforever/actors/net/SocketActor.scala b/src/main/scala/net/psforever/actors/net/SocketActor.scala index 9f3ef4315..09ca2e26d 100644 --- a/src/main/scala/net/psforever/actors/net/SocketActor.scala +++ b/src/main/scala/net/psforever/actors/net/SocketActor.scala @@ -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) => diff --git a/src/main/scala/net/psforever/actors/net/SocketPane.scala b/src/main/scala/net/psforever/actors/net/SocketPane.scala new file mode 100644 index 000000000..c82b05fbf --- /dev/null +++ b/src/main/scala/net/psforever/actors/net/SocketPane.scala @@ -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. + *

+ * 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. + *

+ * 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 + } + } +} diff --git a/src/main/scala/net/psforever/services/account/AccountIntermediaryService.scala b/src/main/scala/net/psforever/services/account/AccountIntermediaryService.scala index 8db0d20a2..3f884bfa6 100644 --- a/src/main/scala/net/psforever/services/account/AccountIntermediaryService.scala +++ b/src/main/scala/net/psforever/services/account/AccountIntermediaryService.scala @@ -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) diff --git a/src/main/scala/net/psforever/util/Config.scala b/src/main/scala/net/psforever/util/Config.scala index bcbcf7f98..419e66cb7 100644 --- a/src/main/scala/net/psforever/util/Config.scala +++ b/src/main/scala/net/psforever/util/Config.scala @@ -104,6 +104,7 @@ case class LoginConfig( case class WorldConfig( port: Int, + ports: Seq[Int], serverName: String, serverType: ServerType )