created a generic base model for automated object deletion that isn't user driven; the first instance is the DroppedItemRemover for LocalService

pslogin/src/main/scala/WorldSessionActor.scala

@@ -37,13 +37,14 @@ import net.psforever.objects.vehicles.{AccessPermissionGroup, Utility, VehicleLo
 import net.psforever.objects.zones.{InterstellarCluster, Zone}
 import net.psforever.packet.game.objectcreate._
 import net.psforever.types._
-import services._
+import services.{RemoverActor, _}
 import services.avatar.{AvatarAction, AvatarResponse, AvatarServiceMessage, AvatarServiceResponse}
 import services.local.{LocalAction, LocalResponse, LocalServiceMessage, LocalServiceResponse}
 import services.vehicle.VehicleAction.UnstowEquipment
 import services.vehicle.{VehicleAction, VehicleResponse, VehicleServiceMessage, VehicleServiceResponse}
 
 import scala.annotation.tailrec
+import scala.concurrent.duration._
 import scala.util.Success
 
 class WorldSessionActor extends Actor with MDCContextAware {
@@ -428,6 +429,11 @@ class WorldSessionActor extends Actor with MDCContextAware {
             sendResponse(HackMessage(0, target_guid, guid, 100, unk1, HackState.Hacked, unk2))
           }
 
+        case LocalResponse.ObjectDelete(item_guid, unk) =>
+          if(tplayer_guid != guid) {
+            sendResponse(ObjectDeleteMessage(item_guid, unk))
+          }
+
         case LocalResponse.ProximityTerminalEffect(object_guid, effectState) =>
           if(tplayer_guid != guid) {
             sendResponse(ProximityTerminalUseMessage(PlanetSideGUID(0), object_guid, effectState))
@@ -590,7 +596,6 @@ class WorldSessionActor extends Actor with MDCContextAware {
         sendResponse(DeployRequestMessage(player.GUID, vehicle_guid, state, 0, false, Vector3.Zero))
         vehicleService ! VehicleServiceMessage(continent.Id, VehicleAction.DeployRequest(player.GUID, vehicle_guid, state, 0, false, Vector3.Zero))
         DeploymentActivities(obj)
-        import scala.concurrent.duration._
         import scala.concurrent.ExecutionContext.Implicits.global
         context.system.scheduler.scheduleOnce(obj.DeployTime milliseconds, obj.Actor, Deployment.TryDeploy(DriveState.Deployed))
       }
@@ -612,7 +617,6 @@ class WorldSessionActor extends Actor with MDCContextAware {
         sendResponse(DeployRequestMessage(player.GUID, vehicle_guid, state, 0, false, Vector3.Zero))
         vehicleService ! VehicleServiceMessage(continent.Id, VehicleAction.DeployRequest(player.GUID, vehicle_guid, state, 0, false, Vector3.Zero))
         DeploymentActivities(obj)
-        import scala.concurrent.duration._
         import scala.concurrent.ExecutionContext.Implicits.global
         context.system.scheduler.scheduleOnce(obj.UndeployTime milliseconds, obj.Actor, Deployment.TryUndeploy(DriveState.Mobile))
       }
@@ -1355,7 +1359,6 @@ class WorldSessionActor extends Actor with MDCContextAware {
           (taskResolver, TaskBeforeZoneChange(GUIDTask.UnregisterAvatar(original)(continent.GUID), zone_id))
         }
       }
-      import scala.concurrent.duration._
       import scala.concurrent.ExecutionContext.Implicits.global
       respawnTimer = context.system.scheduler.scheduleOnce(respawnTime seconds, target, msg)
 
@@ -1387,6 +1390,7 @@ class WorldSessionActor extends Actor with MDCContextAware {
         case None =>
           PlanetSideGUID(0) //object is being introduced into the world upon drop
       }
+      localService ! RemoverActor.AddTask(item, continent, Some(20 seconds))
       localService ! LocalServiceMessage(continent.Id, LocalAction.DropItem(exclusionId, item))
 
     case Zone.Ground.CanNotDropItem(item) =>
@@ -1400,6 +1404,7 @@ class WorldSessionActor extends Actor with MDCContextAware {
     case Zone.Ground.ItemInHand(item) =>
       player.Fit(item) match {
         case Some(slotNum) =>
+          localService ! RemoverActor.ClearSpecific(List(item), continent)
           val item_guid = item.GUID
           val player_guid = player.GUID
           player.Slot(slotNum).Equipment = item
@@ -1476,7 +1481,6 @@ class WorldSessionActor extends Actor with MDCContextAware {
       if(!corpse.isAlive && corpse.HasGUID) {
         corpse.VehicleSeated match {
           case Some(_) =>
-            import scala.concurrent.duration._
             import scala.concurrent.ExecutionContext.Implicits.global
             context.system.scheduler.scheduleOnce(50 milliseconds, self, UnregisterCorpseOnVehicleDisembark(corpse))
           case None =>
@@ -1546,7 +1550,6 @@ class WorldSessionActor extends Actor with MDCContextAware {
         else { //continue next tick
           tickAction.getOrElse(() => Unit)()
           progressBarValue = Some(progressBarVal)
-          import scala.concurrent.duration._
           import scala.concurrent.ExecutionContext.Implicits.global
           progressBarUpdate = context.system.scheduler.scheduleOnce(250 milliseconds, self, ItemHacking(tplayer, target, tool_guid, delta, completeAction))
         }
@@ -1634,7 +1637,6 @@ class WorldSessionActor extends Actor with MDCContextAware {
       player.Locker.Inventory += 0 -> SimpleItem(remote_electronics_kit)
       //TODO end temp player character auto-loading
       self ! ListAccountCharacters
-      import scala.concurrent.duration._
       import scala.concurrent.ExecutionContext.Implicits.global
       clientKeepAlive.cancel
       clientKeepAlive = context.system.scheduler.schedule(0 seconds, 500 milliseconds, self, PokeClient())
@@ -1855,7 +1857,6 @@ class WorldSessionActor extends Actor with MDCContextAware {
           avatarService ! AvatarServiceMessage(continent.Id, AvatarAction.ObjectDelete(player_guid, player_guid, 0))
           self ! PacketCoding.CreateGamePacket(0, DismountVehicleMsg(player_guid, BailType.Normal, true)) //let vehicle try to clean up its fields
 
-          import scala.concurrent.duration._
           import scala.concurrent.ExecutionContext.Implicits.global
           context.system.scheduler.scheduleOnce(50 milliseconds, self, UnregisterCorpseOnVehicleDisembark(player))
           //sendResponse(ObjectDetachMessage(vehicle_guid, player.GUID, Vector3.Zero, 0))
@@ -4299,7 +4300,6 @@ class WorldSessionActor extends Actor with MDCContextAware {
     PlayerActionsToCancel()
     CancelAllProximityUnits()
 
-    import scala.concurrent.duration._
     import scala.concurrent.ExecutionContext.Implicits.global
     reviveTimer = context.system.scheduler.scheduleOnce(respawnTimer milliseconds, galaxy, Zone.Lattice.RequestSpawnPoint(Zones.SanctuaryZoneNumber(tplayer.Faction), tplayer, 7))
   }
@@ -4440,7 +4440,6 @@ class WorldSessionActor extends Actor with MDCContextAware {
     */
   def TryDisposeOfLootedCorpse(obj : Player) : Boolean = {
     if(WellLootedCorpse(obj)) {
-      import scala.concurrent.duration._
       import scala.concurrent.ExecutionContext.Implicits.global
       context.system.scheduler.scheduleOnce(1 second, avatarService, AvatarServiceMessage.RemoveSpecificCorpse(List(obj)))
       true
@@ -4518,7 +4517,6 @@ class WorldSessionActor extends Actor with MDCContextAware {
   def SetDelayedProximityUnitReset(terminal : Terminal with ProximityUnit) : Unit = {
     val terminal_guid = terminal.GUID
     ClearDelayedProximityUnitReset(terminal_guid)
-    import scala.concurrent.duration._
     import scala.concurrent.ExecutionContext.Implicits.global
     delayedProximityTerminalResets += terminal_guid ->
       context.system.scheduler.scheduleOnce(3000 milliseconds, self, DelayedProximityUnitStop(terminal))

pslogin/src/main/scala/services/RemoverActor.scala

@@ -0,0 +1,284 @@
+// Copyright (c) 2017 PSForever
+package services
+
+import akka.actor.{Actor, ActorRef, Cancellable}
+import net.psforever.objects.guid.TaskResolver
+import net.psforever.objects.zones.Zone
+import net.psforever.objects.{DefaultCancellable, PlanetSideGameObject}
+import net.psforever.types.Vector3
+
+import scala.annotation.tailrec
+import scala.concurrent.duration._
+
+abstract class RemoverActor extends Actor {
+  protected var firstTask : Cancellable = DefaultCancellable.obj
+  protected var firstHeap : List[RemoverActor.Entry] = List()
+
+  protected var secondTask : Cancellable = DefaultCancellable.obj
+  protected var secondHeap : List[RemoverActor.Entry] = List()
+
+  protected var taskResolver : ActorRef = Actor.noSender
+
+  protected[this] val log = org.log4s.getLogger
+
+  override def preStart() : Unit = {
+    super.preStart()
+    self ! RemoverActor.Startup()
+  }
+
+  override def postStop() = {
+    super.postStop()
+    firstTask.cancel
+    secondTask.cancel
+
+    firstHeap.foreach(entry => {
+      FirstJob(entry)
+      SecondJob(entry)
+    })
+    secondHeap.foreach { SecondJob }
+  }
+
+  def receive : Receive = {
+    case RemoverActor.Startup() =>
+      ServiceManager.serviceManager ! ServiceManager.Lookup("taskResolver") //ask for a resolver to deal with the GUID system
+
+    case ServiceManager.LookupResult("taskResolver", endpoint) =>
+      taskResolver = endpoint
+      context.become(Processing)
+
+    case msg =>
+      log.error(s"received message $msg before being properly initialized")
+  }
+
+  def Processing : Receive = {
+    case RemoverActor.AddTask(obj, zone, duration) =>
+      val entry = RemoverActor.Entry(obj, zone, duration.getOrElse(FirstStandardDuration).toNanos)
+      if(InclusionTest(entry) && !secondHeap.exists(test => RemoverActor.Similarity(test, entry) )) {
+        InitialJob(entry)
+        if(firstHeap.isEmpty) {
+          //we were the only entry so the event must be started from scratch
+          firstHeap = List(entry)
+          RetimeFirstTask()
+        }
+        else {
+          //unknown number of entries; append, sort, then re-time tasking
+          val oldHead = firstHeap.head
+          if(!firstHeap.exists(test => RemoverActor.Similarity(test, entry))) {
+            firstHeap = (firstHeap :+ entry).sortBy(_.duration)
+            if(oldHead != firstHeap.head) {
+              RetimeFirstTask()
+            }
+          }
+          else {
+            log.trace(s"$obj is already queued for removal")
+          }
+        }
+      }
+      else {
+        log.trace(s"$obj either does not qualify for this Remover or is already queued")
+      }
+
+    case RemoverActor.HurrySpecific(targets, zone) =>
+      CullTargetsFromFirstHeap(targets, zone) match {
+        case Nil => ;
+        case list =>
+          secondTask.cancel
+          list.foreach { FirstJob }
+          secondHeap = list ++ secondHeap
+          import scala.concurrent.ExecutionContext.Implicits.global
+          secondTask = context.system.scheduler.scheduleOnce(SecondStandardDuration, self, RemoverActor.TryDelete())
+      }
+
+    case RemoverActor.HurryAll() =>
+      firstTask.cancel
+      firstHeap.foreach { FirstJob }
+      secondHeap = secondHeap ++ firstHeap
+      firstHeap = Nil
+      secondTask.cancel
+      import scala.concurrent.ExecutionContext.Implicits.global
+      secondTask = context.system.scheduler.scheduleOnce(SecondStandardDuration, self, RemoverActor.TryDelete())
+
+    case RemoverActor.ClearSpecific(targets, zone) =>
+      CullTargetsFromFirstHeap(targets, zone)
+
+    case RemoverActor.ClearAll() =>
+      firstTask.cancel
+      firstHeap = Nil
+
+      //private messages
+    case RemoverActor.StartDelete() =>
+      firstTask.cancel
+      secondTask.cancel
+      val now : Long = System.nanoTime
+      val (in, out) = firstHeap.partition(entry => { now - entry.time >= entry.duration })
+      firstHeap = out
+      secondHeap = secondHeap ++ in
+      in.foreach { FirstJob }
+      RetimeFirstTask()
+      if(secondHeap.nonEmpty) {
+        import scala.concurrent.ExecutionContext.Implicits.global
+        secondTask = context.system.scheduler.scheduleOnce(SecondStandardDuration, self, RemoverActor.TryDelete())
+      }
+      log.trace(s"item removal task has found ${secondHeap.size} items to remove")
+
+    case RemoverActor.TryDelete() =>
+      secondTask.cancel
+      val (in, out) = secondHeap.partition { ClearanceTest }
+      secondHeap = out
+      in.foreach { SecondJob }
+      if(out.nonEmpty) {
+        import scala.concurrent.ExecutionContext.Implicits.global
+        secondTask = context.system.scheduler.scheduleOnce(SecondStandardDuration, self, RemoverActor.TryDelete())
+      }
+      log.trace(s"item removal task has removed ${in.size} items")
+
+    case RemoverActor.FailureToWork(entry, ex) =>
+      log.error(s"${entry.obj} from ${entry.zone} not properly unregistered - $ex")
+  }
+
+  private def CullTargetsFromFirstHeap(targets : List[PlanetSideGameObject], zone : Zone) : List[RemoverActor.Entry] = {
+    if(targets.nonEmpty) {
+      firstTask.cancel
+      val culledEntries = if(targets.size == 1) {
+        log.debug(s"a target submitted for early cleanup: ${targets.head}")
+        //simple selection
+        RemoverActor.recursiveFind(firstHeap.iterator, RemoverActor.Entry(targets.head, zone, 0)) match {
+          case None => ;
+            Nil
+          case Some(index) =>
+            val entry = firstHeap(index)
+            firstHeap = (firstHeap.take(index) ++ firstHeap.drop(index + 1)).sortBy(_.duration)
+            List(entry)
+        }
+      }
+      else {
+        log.trace(s"multiple targets submitted for early cleanup: $targets")
+        //cumbersome partition
+        //a - find targets from entries
+        val locatedTargets = for {
+          a <- targets.map(RemoverActor.Entry(_, zone, 0))
+          b <- firstHeap
+          if b.obj.HasGUID && a.obj.HasGUID && RemoverActor.Similarity(b, a)
+        } yield b
+        if(locatedTargets.nonEmpty) {
+          //b - entries, after the found targets are removed (cull any non-GUID entries while at it)
+          firstHeap = (for {
+            a <- locatedTargets
+            b <- firstHeap
+            if b.obj.HasGUID && a.obj.HasGUID && !RemoverActor.Similarity(b, a)
+          } yield b).sortBy(_.duration)
+          locatedTargets
+        }
+        else {
+          Nil
+        }
+      }
+      RetimeFirstTask()
+      culledEntries
+    }
+    else {
+      Nil
+    }
+  }
+
+  def RetimeFirstTask(now : Long = System.nanoTime) : Unit = {
+    firstTask.cancel
+    if(firstHeap.nonEmpty) {
+      val short_timeout : FiniteDuration = math.max(1, firstHeap.head.duration - (now - firstHeap.head.time)) nanoseconds
+      import scala.concurrent.ExecutionContext.Implicits.global
+      firstTask = context.system.scheduler.scheduleOnce(short_timeout, self, RemoverActor.StartDelete())
+    }
+  }
+
+  def SecondJob(entry : RemoverActor.Entry) : Unit = {
+    entry.obj.Position = Vector3.Zero //somewhere it will not disturb anything
+    taskResolver ! FinalTask(entry)
+  }
+
+  def FinalTask(entry : RemoverActor.Entry) : TaskResolver.GiveTask = {
+    import net.psforever.objects.guid.Task
+    TaskResolver.GiveTask (
+      new Task() {
+        private val localEntry = entry
+        private val localAnnounce = self
+
+        override def isComplete : Task.Resolution.Value = if(!localEntry.obj.HasGUID) {
+          Task.Resolution.Success
+        }
+        else {
+          Task.Resolution.Incomplete
+        }
+
+        def Execute(resolver : ActorRef) : Unit = {
+          resolver ! scala.util.Success(this)
+        }
+
+        override def onFailure(ex : Throwable): Unit = {
+          localAnnounce ! RemoverActor.FailureToWork(localEntry, ex)
+        }
+      }, List(DeletionTask(entry))
+    )
+  }
+
+  def FirstStandardDuration : FiniteDuration
+
+  def SecondStandardDuration : FiniteDuration
+
+  def InclusionTest(entry : RemoverActor.Entry) : Boolean
+
+  def InitialJob(entry : RemoverActor.Entry) : Unit
+
+  def FirstJob(entry : RemoverActor.Entry) : Unit
+
+  def ClearanceTest(entry : RemoverActor.Entry) : Boolean
+
+  def DeletionTask(entry : RemoverActor.Entry) : TaskResolver.GiveTask
+}
+
+object RemoverActor {
+  /**
+    * na
+    * @param obj the target
+    * @param zone the zone in which this target is registered
+    * @param duration how much longer the target will exist (in nanoseconds)
+    * @param time when this entry was created (in nanoseconds)
+    */
+  case class Entry(obj : PlanetSideGameObject, zone : Zone, duration : Long, time : Long = System.nanoTime)
+
+  case class Startup()
+
+  case class AddTask(obj : PlanetSideGameObject, zone : Zone, duration : Option[FiniteDuration] = None)
+
+  case class HurrySpecific(targets : List[PlanetSideGameObject], zone : Zone)
+
+  case class HurryAll()
+
+  case class ClearSpecific(targets : List[PlanetSideGameObject], zone : Zone)
+
+  case class ClearAll()
+
+  protected final case class FailureToWork(entry : RemoverActor.Entry, ex : Throwable)
+
+  private final case class StartDelete()
+
+  private final case class TryDelete()
+
+  private def Similarity(entry1 : RemoverActor.Entry, entry2 : RemoverActor.Entry) : Boolean = {
+    entry1.obj == entry2.obj && entry1.zone == entry2.zone && entry1.obj.GUID == entry2.obj.GUID
+  }
+
+  @tailrec private def recursiveFind(iter : Iterator[RemoverActor.Entry], target : RemoverActor.Entry, index : Int = 0) : Option[Int] = {
+    if(!iter.hasNext) {
+      None
+    }
+    else {
+      val entry = iter.next
+      if(entry.obj.HasGUID && target.obj.HasGUID && Similarity(entry, target)) {
+        Some(index)
+      }
+      else {
+        recursiveFind(iter, target, index + 1)
+      }
+    }
+  }
+}

pslogin/src/main/scala/services/local/LocalAction.scala

@@ -16,6 +16,7 @@ object LocalAction {
   final case class DropItem(player_guid : PlanetSideGUID, item : Equipment) extends Action
   final case class HackClear(player_guid : PlanetSideGUID, target : PlanetSideServerObject, unk1 : Long, unk2 : Long = 8L) extends Action
   final case class HackTemporarily(player_guid : PlanetSideGUID, continent : Zone, target : PlanetSideServerObject, unk1 : Long, unk2 : Long = 8L) extends Action
+  final case class ObjectDelete(player_guid : PlanetSideGUID, item_guid : PlanetSideGUID, unk : Int = 0) extends Action
   final case class ProximityTerminalEffect(player_guid : PlanetSideGUID, object_guid : PlanetSideGUID, effectState : Boolean) extends Action
   final case class TriggerSound(player_guid : PlanetSideGUID, sound : TriggeredSound.Value, pos : Vector3, unk : Int, volume : Float) extends Action
 }

pslogin/src/main/scala/services/local/LocalResponse.scala

@@ -13,6 +13,7 @@ object LocalResponse {
   final case class DropItem(item_id : Int, item_guid : PlanetSideGUID, item_data : ConstructorData) extends Response
   final case class HackClear(target_guid : PlanetSideGUID, unk1 : Long, unk2 : Long) extends Response
   final case class HackObject(target_guid : PlanetSideGUID, unk1 : Long, unk2 : Long) extends Response
+  final case class ObjectDelete(item_guid : PlanetSideGUID, unk : Int) extends Response
   final case class ProximityTerminalEffect(object_guid : PlanetSideGUID, effectState : Boolean) extends Response
   final case class TriggerSound(sound : TriggeredSound.Value, pos : Vector3, unk : Int, volume : Float) extends Response
 }

pslogin/src/main/scala/services/local/LocalService.scala

@@ -3,12 +3,13 @@ package services.local
 
 import akka.actor.{Actor, Props}
 import net.psforever.packet.game.objectcreate.{DroppedItemData, PlacementData}
-import services.local.support.{DoorCloseActor, HackClearActor}
+import services.local.support.{DoorCloseActor, DroppedItemRemover, HackClearActor}
-import services.{GenericEventBus, Service}
+import services.{GenericEventBus, RemoverActor, Service}
 
 class LocalService extends Actor {
   private val doorCloser = context.actorOf(Props[DoorCloseActor], "local-door-closer")
   private val hackClearer = context.actorOf(Props[HackClearActor], "local-hack-clearer")
+  private val janitor = context.actorOf(Props[DroppedItemRemover], "local-item-remover")
   private [this] val log = org.log4s.getLogger
 
   override def preStart = {
@@ -58,6 +59,10 @@ class LocalService extends Actor {
               LocalResponse.DropItem(definition.ObjectId, item.GUID, objectData)
             )
           )
+        case LocalAction.ObjectDelete(player_guid, item_guid, unk) =>
+          LocalEvents.publish(
+            LocalServiceResponse(s"/$forChannel/Local", player_guid, LocalResponse.ObjectDelete(item_guid, unk))
+          )
         case LocalAction.HackClear(player_guid, target, unk1, unk2) =>
           LocalEvents.publish(
             LocalServiceResponse(s"/$forChannel/Local", player_guid, LocalResponse.HackClear(target.GUID, unk1, unk2))
@@ -78,6 +83,12 @@ class LocalService extends Actor {
         case _ => ;
       }
 
+    //messages to DroppedItemRemover
+    case msg @ (RemoverActor.AddTask |
+                RemoverActor.HurrySpecific | RemoverActor.HurryAll |
+                RemoverActor.ClearSpecific | RemoverActor.ClearAll) =>
+      janitor ! msg
+
     //response from DoorCloseActor
     case DoorCloseActor.CloseTheDoor(door_guid, zone_id) =>
       LocalEvents.publish(

pslogin/src/main/scala/services/local/support/DroppedItemRemover.scala

@@ -0,0 +1,33 @@
+// Copyright (c) 2017 PSForever
+package services.local.support
+
+import net.psforever.objects.equipment.Equipment
+import net.psforever.objects.guid.{GUIDTask, TaskResolver}
+import services.{RemoverActor, Service}
+import services.local.{LocalAction, LocalServiceMessage}
+
+import scala.concurrent.duration._
+
+class DroppedItemRemover extends RemoverActor {
+  final val FirstStandardDuration : FiniteDuration = 3 minutes
+
+  final val SecondStandardDuration : FiniteDuration = 500 milliseconds
+
+  def InclusionTest(entry : RemoverActor.Entry) : Boolean = {
+    entry.obj.isInstanceOf[Equipment]
+  }
+
+  def InitialJob(entry : RemoverActor.Entry) : Unit = { }
+
+  def FirstJob(entry : RemoverActor.Entry) : Unit = {
+    import net.psforever.objects.zones.Zone
+    entry.zone.Ground ! Zone.Ground.PickupItem(entry.obj.GUID)
+    context.parent ! LocalServiceMessage(entry.zone.Id, LocalAction.ObjectDelete(Service.defaultPlayerGUID, entry.obj.GUID))
+  }
+
+  def ClearanceTest(entry : RemoverActor.Entry) : Boolean = true
+
+  def DeletionTask(entry : RemoverActor.Entry) : TaskResolver.GiveTask = {
+    GUIDTask.UnregisterEquipment(entry.obj.asInstanceOf[Equipment])(entry.zone.GUID)
+  }
+}