feat: Add Communication Module in Java

Author: harrki

adf_core_python/core/gateway/gateway_agent.py

@@ -2,7 +2,7 @@
 
 from typing import Optional, TYPE_CHECKING, Callable
 
-from rcrs_core.connection import RCRSProto_pb2
+from rcrs_core.connection import RCRSProto_pb2, URN
 
 from adf_core_python.core.agent.info.agent_info import AgentInfo
 from adf_core_python.core.agent.info.scenario_info import ScenarioInfo
@@ -111,3 +111,8 @@ def message_received(self, msg: RCRSProto_pb2) -> None:
 
             self._gateway_modules[c_msg.module_id].set_execute_response(c_msg.result)
             self._gateway_modules[c_msg.module_id].set_is_executed(True)
+
+        if msg.urn == URN.Command.AK_SPEAK:
+            if self.send_msg is None:
+                raise RuntimeError("send_msg is None")
+            self.send_msg(msg)

adf_core_python/launcher.py

@@ -1,4 +1,5 @@
 import argparse
+import resource
 
 from adf_core_python.core.config.config import Config
 from adf_core_python.core.launcher.agent_launcher import AgentLauncher
@@ -11,6 +12,8 @@ def __init__(
         self,
         launcher_config_file: str,
     ) -> None:
+        resource.setrlimit(resource.RLIMIT_NOFILE, (8192, 9223372036854775807))
+
         configure_logger()
 
         self.logger = get_logger(__name__)

java/lib/src/main/java/adf_core_python/agent/Agent.java

@@ -1,14 +1,19 @@
 package adf_core_python.agent;
 
-import adf.core.agent.communication.MessageManager;
+import adf.core.agent.communication.standard.bundle.StandardMessageBundle;
 import adf.core.agent.info.ScenarioInfo;
 import adf.core.agent.info.WorldInfo;
+import adf.core.launcher.ConsoleOutput;
+import adf_core_python.agent.communication.MessageManager;
+import adf_core_python.agent.communication.standard.StandardCommunicationModule;
 import adf_core_python.agent.config.ModuleConfig;
 import adf_core_python.agent.develop.DevelopData;
 import adf_core_python.agent.info.AgentInfo;
 import adf_core_python.agent.module.ModuleManager;
 import adf_core_python.agent.precompute.PrecomputeData;
+import adf_core_python.component.communication.CommunicationModule;
 import adf_core_python.component.module.AbstractModule;
+import adf_core_python.gateway.Coordinator;
 import adf_core_python.gateway.mapper.AbstractMapper;
 import adf_core_python.gateway.mapper.MapperDict;
 import jakarta.annotation.Nonnull;
@@ -17,40 +22,50 @@
 import org.apache.logging.log4j.Logger;
 import rescuecore2.config.Config;
 import rescuecore2.messages.Command;
+import rescuecore2.messages.Message;
 import rescuecore2.standard.entities.StandardEntityURN;
 import rescuecore2.standard.entities.StandardWorldModel;
+import rescuecore2.standard.messages.AKSubscribe;
 import rescuecore2.worldmodel.ChangeSet;
 import rescuecore2.worldmodel.Entity;
 import rescuecore2.worldmodel.EntityID;
 
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Objects;
 
 public class Agent {
-    private final AgentInfo agentInfo;
-    private final WorldInfo worldInfo;
-    private final ScenarioInfo scenarioInfo;
+    public final AgentInfo agentInfo;
+    public final WorldInfo worldInfo;
+    public final ScenarioInfo scenarioInfo;
     private final ModuleManager moduleManager;
     private final DevelopData developData;
     private final PrecomputeData precomputeData;
     private final MessageManager messageManager;
+    private CommunicationModule communicationModule;
     private final HashMap<String, AbstractMapper> modules = new HashMap<>();
     private final MapperDict mapperDict;
     private final Logger logger;
+    private int ignoreTime;
+    private final Coordinator coordinator;
 
-    public Agent(EntityID entityID, Collection<Entity> entities, ScenarioInfo scenarioInfo, DevelopData developData, ModuleConfig moduleConfig) {
+    public Agent(EntityID entityID, Collection<Entity> entities, ScenarioInfo scenarioInfo, DevelopData developData, ModuleConfig moduleConfig, Coordinator coordinator) {
         StandardWorldModel worldModel = new StandardWorldModel();
         worldModel.addEntities(entities);
         worldModel.index();
 
+        this.ignoreTime = scenarioInfo.getRawConfig()
+                .getIntValue(kernel.KernelConstants.IGNORE_AGENT_COMMANDS_KEY);
+
         this.agentInfo = new AgentInfo(entityID, worldModel);
         this.worldInfo = new WorldInfo(worldModel);
         this.scenarioInfo = scenarioInfo;
         this.developData = developData;
         this.moduleManager = new ModuleManager(this.agentInfo, this.worldInfo, this.scenarioInfo, moduleConfig, this.developData);
+        this.coordinator = coordinator;
 
         String dataStorageName = "";
         StandardEntityURN agentURN = Objects.requireNonNull(this.worldInfo.getEntity(this.agentInfo.getID())).getStandardURN();
@@ -97,13 +112,48 @@ public Class<?> registerModule(@Nonnull String moduleID, @Nonnull String moduleN
     }
 
     public void update(int time, ChangeSet changed, Collection<Command> heard) {
+        worldInfo.setTime(time);
+        worldInfo.merge(changed);
         agentInfo.recordThinkStartTime();
         agentInfo.setTime(time);
+
+        if (time == 1) {
+            if (this.communicationModule != null) {
+                ConsoleOutput.out(ConsoleOutput.State.ERROR,
+                        "[ERROR ] Loader is not found.");
+                ConsoleOutput.out(ConsoleOutput.State.NOTICE,
+                        "CommunicationModule is modified - " + this);
+            } else {
+                this.communicationModule = new StandardCommunicationModule();
+            }
+
+            this.messageManager.registerMessageBundle(new StandardMessageBundle());
+        }
+
+        // agents can subscribe after ignore time
+        if (time >= ignoreTime) {
+            this.messageManager.subscribe(this.agentInfo, this.worldInfo,
+                    this.scenarioInfo);
+
+            if (!this.messageManager.getIsSubscribed()) {
+                int[] channelsToSubscribe = this.messageManager.getChannels();
+                if (channelsToSubscribe != null) {
+                    this.messageManager.setIsSubscribed(true);
+                }
+            }
+        }
+
         agentInfo.setHeard(heard);
         agentInfo.setChanged(changed);
-        worldInfo.setTime(time);
-        worldInfo.merge(changed);
         worldInfo.setChanged(changed);
+
+        this.messageManager.refresh();
+        this.communicationModule.receive(this, this.messageManager);
+
+        this.messageManager.coordinateMessages(this.agentInfo, this.worldInfo,
+                this.scenarioInfo);
+        this.communicationModule.send(this, this.messageManager);
+
         logger.debug("Agent Update (Time: {}, Changed: {}, Heard: {})", agentInfo.getTime(), agentInfo.getChanged(), agentInfo.getHeard());
     }
 
@@ -113,4 +163,12 @@ public Config execModuleMethod(String moduleID, String methodName, Config argume
         logger.debug("Executed Method Result (MethodName: {}, Result: {}", methodName, result);
         return result;
     }
+
+    public EntityID getID() {
+        return this.agentInfo.getID();
+    }
+
+    public void send(Message[] messages) {
+        Arrays.stream(messages).forEach(coordinator::sendMessage);
+    }
 }