IPC Channel ID's (supports subscribing to events from a single broker), more work on EditorMode/PlayerMode IPC communication (JSErrors)

Source/Atomic/IPC/IPC.cpp

@@ -15,7 +15,8 @@
 namespace Atomic
 {
 
-IPC::IPC(Context* context) : Object(context)
+IPC::IPC(Context* context) : Object(context),
+    workerChannelID_(0)
 {
     SubscribeToEvent(E_UPDATE, HANDLER(IPC, HandleUpdate));
 }
@@ -33,15 +34,16 @@ IPC::~IPC()
     worker_ = 0;
 }
 
-bool IPC::InitWorker(IPCHandle fd1, IPCHandle fd2)
+bool IPC::InitWorker(unsigned id, IPCHandle fd1, IPCHandle fd2)
 {
+    workerChannelID_ = id;
 
 #ifndef ATOMIC_PLATFORM_WINDOWS
     // close server fd
     close(fd1);
 #endif
 
-    worker_ = new IPCWorker(fd2, context_);
+    worker_ = new IPCWorker(context_, fd2, id);
     worker_->Run();
 
     SendEventToBroker(E_IPCWORKERSTART);
@@ -97,8 +99,8 @@ void IPC::HandleUpdate(StringHash eventType, VariantMap& eventData)
         if (!broker->Update())
         {
             VariantMap brokerData;
-            brokerData[WorkerExit::P_BROKER] = broker;
-            brokerData[WorkerExit::P_EXITCODE] = 0;
+            brokerData[IPCWorkerExit::P_BROKER] = broker;
+            brokerData[IPCWorkerExit::P_EXITCODE] = 0;
             broker->SendEvent(E_IPCWORKEREXIT, brokerData);
             remove.Push(broker);
         }
@@ -113,9 +115,27 @@ void IPC::HandleUpdate(StringHash eventType, VariantMap& eventData)
 
     for (List<QueuedEvent>::Iterator itr = queuedEvents_.Begin(); itr != queuedEvents_.End(); itr++)
     {
+        unsigned channelID = (*itr).channelID_;
         StringHash qeventType =  (*itr).eventType_;
         VariantMap& qeventData =  (*itr).eventData_;
-        SendEvent(qeventType, qeventData);
+
+        bool sent = false;
+
+        for (unsigned i = 0; i < brokers_.Size(); i++)
+        {
+            SharedPtr<IPCBroker>& broker = brokers_[i];
+            if (broker->GetID() == channelID) {
+
+                broker->SendEvent(qeventType, qeventData);
+                sent = true;
+                break;
+            }
+        }
+
+        if (!sent)
+            SendEvent(qeventType, qeventData);
+
+
     }
 
     queuedEvents_.Clear();
@@ -123,11 +143,12 @@ void IPC::HandleUpdate(StringHash eventType, VariantMap& eventData)
     eventMutex_.Release();
 }
 
-void IPC::QueueEvent(StringHash eventType, VariantMap& eventData)
+void IPC::QueueEvent(unsigned id, StringHash eventType, VariantMap& eventData)
 {
     eventMutex_.Acquire();
 
     QueuedEvent event;
+    event.channelID_ = id;
     event.eventType_ = eventType;
     event.eventData_ = eventData;
     queuedEvents_.Push(event);

Source/Atomic/IPC/IPC.h

@@ -14,6 +14,7 @@ class IPCWorker;
 
 struct QueuedEvent
 {
+    unsigned channelID_;
     StringHash eventType_;
     VariantMap eventData_;
 };
@@ -30,10 +31,10 @@ public:
     virtual ~IPC();
 
     // queues an event from a worker or broker receiving thread
-    void QueueEvent(StringHash eventType, VariantMap& eventData);
+    void QueueEvent(unsigned id, StringHash eventType, VariantMap& eventData);
 
     // for a child worker process
-    bool InitWorker(IPCHandle fd1, IPCHandle fd2);
+    bool InitWorker(unsigned id, IPCHandle fd1, IPCHandle fd2);
 
     // spawn a worker process
     IPCBroker* SpawnWorker(const String& command, const Vector<String>& args, const String& initialDirectory = "");
@@ -44,6 +45,9 @@ public:
 
 private:
 
+    // if non-zero we're a worked and this is out broker's channel id
+    unsigned workerChannelID_;
+
     // processes queued events
     void HandleUpdate(StringHash eventType, VariantMap& eventData);
 

Source/Atomic/IPC/IPCBroker.cpp

@@ -13,9 +13,11 @@
 namespace Atomic
 {
 
-IPCBroker::IPCBroker(Context* context) : IPCChannel(context)
-{    
+unsigned IPCBroker::idCounter_ = 1;
 
+IPCBroker::IPCBroker(Context* context) : IPCChannel(context, idCounter_)
+{    
+    idCounter_++;
 }
 
 IPCBroker::~IPCBroker()
@@ -93,9 +95,11 @@ bool IPCBroker::SpawnWorker(const String& command, const Vector<String>& args, c
 
 #else
     pargs.Push(ToString("--ipc-server=%i", pp_.fd1()));
-    pargs.Push(ToString("--ipc-client=%i", pp_.fd2()));
+    pargs.Push(ToString("--ipc-client=%i", pp_.fd2()));    
 #endif
 
+    pargs.Push(ToString("--ipc-id=%i", id_));
+
     if (!otherProcess_->Launch(command, pargs, initialDirectory))
         return false;
 

Source/Atomic/IPC/IPCBroker.h

@@ -8,6 +8,8 @@ namespace Atomic
 
 class IPCProcess;
 
+
+/// Server application held broker
 class IPCBroker : public IPCChannel
 {
     friend class IPC;
@@ -26,6 +28,8 @@ public:
 
 private:
 
+    static unsigned idCounter_;
+
     bool SpawnWorker(const String& command, const Vector<String>& args, const String& initialDirectory = "");
 
     // broker instantiates the pipe pair

Source/Atomic/IPC/IPCChannel.cpp

@@ -6,7 +6,8 @@
 namespace Atomic
 {
 
-IPCChannel::IPCChannel(Context* context) : Object(context)
+IPCChannel::IPCChannel(Context* context, unsigned id) : Object(context),
+    id_(id)
 {
     ipc_ = GetSubsystem<IPC>();
     currentHeader_.messageType_ = IPC_MESSAGE_UNDEFINED;
@@ -20,9 +21,7 @@ IPCChannel::~IPCChannel()
 void IPCChannel::PostMessage(StringHash eventType, VariantMap &eventData)
 {
     IPCMessageEvent msgEvent;
-    msgEvent.DoSend(transport_, eventType, eventData);
-
-    LOGERRORF("Posting Message");
+    msgEvent.DoSend(transport_, id_, eventType, eventData);
 }
 
 bool IPCChannel::Receive()
@@ -65,8 +64,9 @@ bool IPCChannel::Receive()
             IPCMessageEvent event;
             StringHash eventType;
             VariantMap eventData;
-            event.DoRead(buffer, eventType, eventData);
-            ipc_->QueueEvent(eventType, eventData);
+            unsigned id;
+            event.DoRead(buffer, id, eventType, eventData);
+            ipc_->QueueEvent(id, eventType, eventData);
         }
 
         if (dataBuffer_.IsEof())

Source/Atomic/IPC/IPCChannel.h

@@ -26,17 +26,21 @@ class IPCChannel : public Object, public Thread
 
 public:
 
-    IPCChannel(Context* context);
+    IPCChannel(Context* context, unsigned id);
     virtual ~IPCChannel();
 
     virtual void ThreadFunction() {}
 
+    unsigned GetID() { return id_; }
+
     bool Receive();
 
     void PostMessage(StringHash eventType, VariantMap& eventData);
 
 protected:
 
+    unsigned id_;
+
     // for access from thread
     WeakPtr<IPC> ipc_;
 

Source/Atomic/IPC/IPCEvents.h

@@ -6,23 +6,29 @@ namespace Atomic
 {
 
 /// Worker start
-EVENT(E_IPCWORKERSTART, WorkerStart)
+EVENT(E_IPCWORKERSTART, IPCWorkerStart)
 {
 
 }
 
 /// Worker exited
-EVENT(E_IPCWORKEREXIT, WorkerExit)
+EVENT(E_IPCWORKEREXIT, IPCWorkerExit)
 {
     PARAM(P_BROKER, Broker);   // Broker*
     PARAM(P_EXITCODE, ExitCode);   // int
 }
 
-/// Worker start
-EVENT(E_IPCHELLOFROMBROKER, HelloFromBroker)
+/// broker -> worker start
+EVENT(E_IPCINITIALIZE, IPCInitialize)
+{
+
+}
+
+/// Worker Log
+EVENT(E_IPCWORKERLOG, IPCWorkerLog)
 {
-    PARAM(P_HELLO, Hello);      // string
-    PARAM(P_LIFETHEUNIVERSEANDEVERYTHING, LifeTheUniverseAndEverything); // 42
+    PARAM(P_LEVEL, Level);      // int log level
+    PARAM(P_MESSAGE, Message);  // string
 }
 
 

Source/Atomic/IPC/IPCMessage.h

@@ -28,17 +28,19 @@ class IPCMessageEvent
 
 public:
 
-    bool DoRead(MemoryBuffer& buffer, StringHash& eventType, VariantMap& eventData)
+    bool DoRead(MemoryBuffer& buffer, unsigned& id, StringHash& eventType, VariantMap& eventData)
     {
+        id = buffer.ReadUInt();
         eventType = buffer.ReadStringHash();
         eventData.Clear();
         eventData = buffer.ReadVariantMap();
         return true;
     }
 
-    bool DoSend(PipeTransport& transport, const StringHash& eventType, const VariantMap& eventData)
+    bool DoSend(PipeTransport& transport, unsigned id, const StringHash& eventType, const VariantMap& eventData)
     {
         VectorBuffer buffer;
+        buffer.WriteUInt(id);
         buffer.WriteStringHash(eventType);
         buffer.WriteVariantMap(eventData);
 

Source/Atomic/IPC/IPCWorker.cpp

@@ -15,7 +15,7 @@
 namespace Atomic
 {
 
-IPCWorker::IPCWorker(IPCHandle fd, Context* context) : IPCChannel(context),
+IPCWorker::IPCWorker(Context* context, IPCHandle fd, unsigned id) : IPCChannel(context, id),
     fd_(fd)
 {
 

Source/Atomic/IPC/IPCWorker.h

@@ -14,7 +14,7 @@ class IPCWorker : public IPCChannel
 
 public:
     /// Construct.
-    IPCWorker(IPCHandle fd, Context* context);
+    IPCWorker(Context* context, IPCHandle fd, unsigned id);
     /// Destruct.
     virtual ~IPCWorker();
 

Source/AtomicEditor/Source/Player/AEPlayer.cpp

@@ -59,10 +59,8 @@ void AEPlayer::HandleJSError(StringHash eventType, VariantMap& eventData)
 
 void AEPlayer::HandleIPCWorkerStarted(StringHash eventType, VariantMap& eventData)
 {
-    VariantMap weventData;
-    weventData[HelloFromBroker::P_HELLO] = "Hello";
-    weventData[HelloFromBroker::P_LIFETHEUNIVERSEANDEVERYTHING] = 42;
-    broker_->PostMessage(E_IPCHELLOFROMBROKER, weventData);
+    //VariantMap weventData;
+    //broker_->PostMessage(E_IPCSTARTUP, weventData);
 }
 
 void AEPlayer::HandleIPCWorkerExit(StringHash eventType, VariantMap& eventData)

Source/AtomicEditor/Source/Player/AEPlayerApplication.cpp

@@ -179,8 +179,8 @@ void AEPlayerApplication::Setup()
 
 void AEPlayerApplication::HandleHelloFromBroker(StringHash eventType, VariantMap& eventData)
 {
-    assert(eventData[HelloFromBroker::P_HELLO].GetString() == "Hello");
-    assert(eventData[HelloFromBroker::P_LIFETHEUNIVERSEANDEVERYTHING].GetInt() == 42);
+    //assert(eventData[HelloFromBroker::P_HELLO].GetString() == "Hello");
+    //assert(eventData[HelloFromBroker::P_LIFETHEUNIVERSEANDEVERYTHING].GetInt() == 42);
 
     LOGERROR("Passed Test!");
 }
@@ -188,7 +188,7 @@ void AEPlayerApplication::HandleHelloFromBroker(StringHash eventType, VariantMap
 void AEPlayerApplication::Start()
 {
 
-    SubscribeToEvent(E_IPCHELLOFROMBROKER, HANDLER(AEPlayerApplication, HandleHelloFromBroker));
+    //SubscribeToEvent(E_IPCHELLOFROMBROKER, HANDLER(AEPlayerApplication, HandleHelloFromBroker));
     SubscribeToEvent(E_JSERROR, HANDLER(AEPlayerApplication, HandleJSError));
 
 
@@ -210,7 +210,7 @@ void AEPlayerApplication::Start()
     {
         IPC* ipc = new IPC(context_);
         context_->RegisterSubsystem(ipc);
-        ipc->InitWorker(fd_[0], fd_[1]);
+        //ipc->InitWorker(fd_[0], fd_[1]);
     }
 #endif
 

Source/AtomicEditorWork/Application/AEPlayerApp.cpp

@@ -145,11 +145,6 @@ void AEPlayerApplication::Start()
     context_->RegisterSubsystem(new AtomicPlayer::Player(context_));
     AtomicPlayer::jsapi_init_atomicplayer(vm_);
 
-    if (!vm_->ExecuteMain())
-    {
-        ErrorExit("Error executing Scripts/main.js");
-    }
-
     return;
 }
 
@@ -197,16 +192,14 @@ void AEPlayerApplication::HandleJSError(StringHash eventType, VariantMap& eventD
 {
     using namespace JSError;
     //String errName = eventData[P_ERRORNAME].GetString();
+    //String errStack = eventData[P_ERRORSTACK].GetString();
     String errMessage = eventData[P_ERRORMESSAGE].GetString();
     String errFilename = eventData[P_ERRORFILENAME].GetString();
-    //String errStack = eventData[P_ERRORSTACK].GetString();
     int errLineNumber = eventData[P_ERRORLINENUMBER].GetInt();
 
     String errorString = ToString("%s - %s - Line: %i",
                                   errFilename.CString(), errMessage.CString(), errLineNumber);
 
-    ErrorExit(errorString);
-
 }
 
 

Source/AtomicEditorWork/EditorMode/AEEditorMode.cpp

@@ -9,6 +9,8 @@
 #include <ToolCore/ToolSystem.h>
 #include <ToolCore/Project/Project.h>
 
+#include <AtomicJS/Javascript/JSIPCEvents.h>
+
 #include "AEEditorMode.h"
 
 using namespace ToolCore;
@@ -29,10 +31,8 @@ EditorMode::~EditorMode()
 
 void EditorMode::HandleIPCWorkerStarted(StringHash eventType, VariantMap& eventData)
 {
-    VariantMap weventData;
-    weventData[HelloFromBroker::P_HELLO] = "Hello";
-    weventData[HelloFromBroker::P_LIFETHEUNIVERSEANDEVERYTHING] = 42;
-    playerBroker_->PostMessage(E_IPCHELLOFROMBROKER, weventData);
+    VariantMap startupData;
+    playerBroker_->PostMessage(E_IPCINITIALIZE, startupData);
 }
 
 void EditorMode::HandleIPCWorkerExit(StringHash eventType, VariantMap& eventData)
@@ -40,6 +40,19 @@ void EditorMode::HandleIPCWorkerExit(StringHash eventType, VariantMap& eventData
     //SendEvent(E_EDITORPLAYSTOP);
 }
 
+void EditorMode::HandleIPCWorkerLog(StringHash eventType, VariantMap& eventData)
+{
+    using namespace IPCWorkerLog;
+    const String&  message = eventData[P_MESSAGE].GetString();
+
+    LOGINFOF("From Player: %s", message.CString());
+
+}
+
+void EditorMode::HandleIPCJSError(StringHash eventType, VariantMap& eventData)
+{
+
+}
 
 bool EditorMode::PlayProject()
 {
@@ -77,8 +90,9 @@ bool EditorMode::PlayProject()
 
     if (playerBroker_)
     {        
+        SubscribeToEvent(playerBroker_, E_IPCJSERROR, HANDLER(EditorMode, HandleIPCJSError));
         SubscribeToEvent(playerBroker_, E_IPCWORKEREXIT, HANDLER(EditorMode, HandleIPCWorkerExit));
-
+        SubscribeToEvent(playerBroker_, E_IPCWORKERLOG, HANDLER(EditorMode, HandleIPCWorkerLog));
     }
 
     return playerBroker_.NotNull();

Source/AtomicEditorWork/EditorMode/AEEditorMode.h

@@ -29,6 +29,8 @@ public:
 private:
 
     void HandleIPCWorkerStarted(StringHash eventType, VariantMap& eventData);
+    void HandleIPCJSError(StringHash eventType, VariantMap& eventData);
+    void HandleIPCWorkerLog(StringHash eventType, VariantMap& eventData);
     void HandleIPCWorkerExit(StringHash eventType, VariantMap& eventData);
 
     SharedPtr<IPCBroker> playerBroker_;

Source/AtomicEditorWork/PlayerMode/AEPlayerMode.cpp

@@ -1,42 +1,73 @@
 
+#include <Atomic/IO/IOEvents.h>
 #include <Atomic/IO/Log.h>
+#include <Atomic/Input/InputEvents.h>
 #include <Atomic/Core/ProcessUtils.h>
-#include <Atomic/IPC/IPC.h>
 #include <Atomic/IPC/IPCEvents.h>
 #include <Atomic/IPC/IPCWorker.h>
 
+#include <AtomicJS/Javascript/JSVM.h>
+#include <AtomicJS/Javascript/JSEvents.h>
+#include <AtomicJS/Javascript/JSIPCEvents.h>
+
 #include "AEPlayerMode.h"
 
 namespace AtomicEditor
 {
 
 PlayerMode::PlayerMode(Context* context) :
-    Object(context)
+    Object(context),
+    brokerActive_(false)
 {
     fd_[0] = INVALID_IPCHANDLE_VALUE;
     fd_[1] = INVALID_IPCHANDLE_VALUE;
+
+    ipc_ = GetSubsystem<IPC>();
+
+    SubscribeToEvent(E_LOGMESSAGE, HANDLER(PlayerMode, HandleLogMessage));
+    SubscribeToEvent(E_JSERROR, HANDLER(PlayerMode, HandleJSError));
 }
 
-void PlayerMode::HandleHelloFromBroker(StringHash eventType, VariantMap& eventData)
+PlayerMode::~PlayerMode()
 {
-    assert(eventData[HelloFromBroker::P_HELLO].GetString() == "Hello");
-    assert(eventData[HelloFromBroker::P_LIFETHEUNIVERSEANDEVERYTHING].GetInt() == 42);
 
-    LOGERROR("Passed Test!");
+}
+
+void PlayerMode::HandleIPCInitialize(StringHash eventType, VariantMap& eventData)
+{
+    brokerActive_ = true;
+
+    JSVM* vm = JSVM::GetJSVM(0);
+
+    if (!vm->ExecuteMain())
+    {
+        //SendEvent(E_EXITREQUESTED);
+    }
+
 }
 
 void PlayerMode::ProcessArguments() {
 
     const Vector<String>& arguments = GetArguments();
 
+    int id = -1;
+
     for (unsigned i = 0; i < arguments.Size(); ++i)
     {
         if (arguments[i].Length() > 1)
         {
             String argument = arguments[i].ToLower();
-            String value = i + 1 < arguments.Size() ? arguments[i + 1] : String::EMPTY;
+            // String value = i + 1 < arguments.Size() ? arguments[i + 1] : String::EMPTY;
 
-            if (argument.StartsWith("--ipc-server=") || argument.StartsWith("--ipc-client="))
+            if (argument.StartsWith("--ipc-id="))
+            {
+                Vector<String> idc = argument.Split(argument.CString(), '=');
+                if (idc.Size() == 2)
+
+                id = ToInt(idc[1].CString());
+            }
+
+            else if (argument.StartsWith("--ipc-server=") || argument.StartsWith("--ipc-client="))
             {
                 LOGINFOF("Starting IPCWorker %s", argument.CString());
 
@@ -88,19 +119,61 @@ void PlayerMode::ProcessArguments() {
         //ipc->InitWorker(fd_[0], fd_[1]);
     }
 #else
-    if (fd_[0] != INVALID_IPCHANDLE_VALUE && fd_[1] != INVALID_IPCHANDLE_VALUE)
-    {
-        IPC* ipc = GetSubsystem<IPC>();
-        SubscribeToEvent(E_IPCHELLOFROMBROKER, HANDLER(PlayerMode, HandleHelloFromBroker));
-        ipc->InitWorker(fd_[0], fd_[1]);
+    if (id > 0 && fd_[0] != INVALID_IPCHANDLE_VALUE && fd_[1] != INVALID_IPCHANDLE_VALUE)
+    {        
+        SubscribeToEvent(E_IPCINITIALIZE, HANDLER(PlayerMode, HandleIPCInitialize));
+        ipc_->InitWorker((unsigned) id, fd_[0], fd_[1]);
     }
 #endif
 
 }
 
-PlayerMode::~PlayerMode()
+void PlayerMode::HandleJSError(StringHash eventType, VariantMap& eventData)
 {
+    if (brokerActive_)
+    {
+        if (ipc_.Null())
+            return;
+
+        String errName = eventData[JSError::P_ERRORNAME].GetString();
+        String errStack = eventData[JSError::P_ERRORSTACK].GetString();
+        String errMessage = eventData[JSError::P_ERRORMESSAGE].GetString();
+        String errFilename = eventData[JSError::P_ERRORFILENAME].GetString();
+        int errLineNumber = eventData[JSError::P_ERRORLINENUMBER].GetInt();
+
+        VariantMap ipcErrorData;
+        ipcErrorData[IPCJSError::P_ERRORNAME] = errName;
+        ipcErrorData[IPCJSError::P_ERRORSTACK] = errStack;
+        ipcErrorData[IPCJSError::P_ERRORMESSAGE] = errMessage;
+        ipcErrorData[IPCJSError::P_ERRORFILENAME] = errFilename;
+        ipcErrorData[IPCJSError::P_ERRORLINENUMBER] = errLineNumber;
+
+        ipc_->SendEventToBroker(E_IPCJSERROR, ipcErrorData);
+
+        LOGERROR("SENDING E_IPCJSERROR");
+
+    }
+
+}
+
+
+void PlayerMode::HandleLogMessage(StringHash eventType, VariantMap& eventData)
+{
+    using namespace LogMessage;
+
+    if (brokerActive_)
+    {
+
+        if (ipc_.Null())
+            return;
+
+        VariantMap logEvent;
+        logEvent[IPCWorkerLog::P_LEVEL] = eventData[P_LEVEL].GetInt();
+        logEvent[IPCWorkerLog::P_MESSAGE] = eventData[P_MESSAGE].GetString();
+        ipc_->SendEventToBroker(E_IPCWORKERLOG, logEvent);
+    }
 
 }
 
+
 }

Source/AtomicEditorWork/PlayerMode/AEPlayerMode.h

@@ -2,6 +2,7 @@
 #pragma once
 
 #include <Atomic/Core/Object.h>
+#include <Atomic/IPC/IPC.h>
 #include <Atomic/IPC/IPCTypes.h>
 
 using namespace Atomic;
@@ -25,10 +26,16 @@ public:
 private:
 
     void ProcessArguments();
-    void HandleHelloFromBroker(StringHash eventType, VariantMap& eventData);
+
+    void HandleJSError(StringHash eventType, VariantMap& eventData);
+    void HandleLogMessage(StringHash eventType, VariantMap& eventData);
+    void HandleIPCInitialize(StringHash eventType, VariantMap& eventData);
 
     IPCHandle fd_[2];
 
+    WeakPtr<IPC> ipc_;
+    bool brokerActive_;
+
 };
 
 }

Source/AtomicJS/Javascript/JSIPCEvents.h

@@ -0,0 +1,21 @@
+// Copyright (c) 2014-2015, THUNDERBEAST GAMES LLC All rights reserved
+// Please see LICENSE.md in repository root for license information
+// https://github.com/AtomicGameEngine/AtomicGameEngine
+
+#pragma once
+
+#include <Atomic/Core/Object.h>
+
+namespace Atomic
+{
+
+EVENT(E_IPCJSERROR, IPCJSError)
+{
+    PARAM(P_ERRORNAME, ErrorName); // string
+    PARAM(P_ERRORMESSAGE, ErrorMessage); // string
+    PARAM(P_ERRORFILENAME, ErrorFileName); // string
+    PARAM(P_ERRORLINENUMBER, ErrorLineNumber); // int
+    PARAM(P_ERRORSTACK, ErrorStack); // string
+}
+
+}