Merge pull request #20 from thomasjm/master

Fix handling of multiple ZMQ identifiers

Author: castarco

src/Actions/Action.php

@@ -14,5 +14,5 @@
 
 interface Action
 {
-    public function call(array $header, array $content, $zmqId = null);
+    public function call(array $header, array $content, $zmqIds = []);
 }

src/Actions/ExecuteAction.php

@@ -57,7 +57,7 @@ public function __construct(
         $this->shellSoul = $shellSoul;
     }
 
-    public function call(array $header, array $content, $zmqId = null)
+    public function call(array $header, array $content, $zmqIds = [])
     {
         $this->broker->send($this->iopubSocket, 'status', ['execution_state' => 'busy'], $header);
 
@@ -85,7 +85,7 @@ public function call(array $header, array $content, $zmqId = null)
             'user_expressions' => new \stdClass
         ];
 
-        $this->broker->send($this->shellSocket, 'execute_reply', $replyContent, $this->header, [], $zmqId);
+        $this->broker->send($this->shellSocket, 'execute_reply', $replyContent, $this->header, [], $zmqIds);
 
         $this->broker->send($this->iopubSocket, 'status', ['execution_state' => 'idle'], $this->header);
     }

src/Actions/HistoryAction.php

@@ -31,8 +31,8 @@ public function __construct(JupyterBroker $broker, SocketWrapper $shellSocket)
         $this->shellSocket = $shellSocket;
     }
 
-    public function call(array $header, array $content, $zmqId = null)
+    public function call(array $header, array $content, $zmqIds = [])
     {
-        $this->broker->send($this->shellSocket, 'history_reply', ['history' => []], $header, [], $zmqId);
+        $this->broker->send($this->shellSocket, 'history_reply', ['history' => []], $header, [], $zmqIds);
     }
 }

src/Actions/KernelInfoAction.php

@@ -33,7 +33,7 @@ public function __construct(JupyterBroker $broker, SocketWrapper $shellSocket, S
         $this->iopubSocket = $iopubSocket;
     }
 
-    public function call(array $header, array $content, $zmqId = null)
+    public function call(array $header, array $content, $zmqIds = [])
     {
         $this->broker->send($this->iopubSocket, 'status', ['execution_state' => 'busy'], $header);
 
@@ -56,7 +56,7 @@ public function call(array $header, array $content, $zmqId = null)
             ],
             $header,
             [],
-            $zmqId
+            $zmqIds
         );
 
         $this->broker->send($this->iopubSocket, 'status', ['execution_state' => 'idle'], $header);

src/Actions/ShutdownAction.php

@@ -32,12 +32,12 @@ public function __construct(JupyterBroker $broker, SocketWrapper $iopubSocket, S
         $this->shellSocket = $shellSocket;
     }
 
-    public function call(array $header, array $content, $zmqId = null)
+    public function call(array $header, array $content, $zmqIds = [])
     {
         $this->broker->send($this->iopubSocket, 'status', ['execution_state' => 'busy'], $header);
 
         $replyContent = ['restart' => $content['restart']];
-        $this->broker->send($this->shellSocket, 'shutdown_reply', $replyContent, $header, [], $zmqId);
+        $this->broker->send($this->shellSocket, 'shutdown_reply', $replyContent, $header, [], $zmqIds);
 
         $this->broker->send($this->iopubSocket, 'status', ['execution_state' => 'idle'], $header);
     }

src/Handlers/ShellMessagesHandler.php

@@ -67,15 +67,23 @@ public function __construct(
 
     public function __invoke(array $msg)
     {
-        list($zmqId, $delim, $hmac, $header, $parentHeader, $metadata, $content) = $msg;
+        // Read ZMQ IDs until we reach the delimiter
+        $zmqIds = array();
+        while (!empty($msg)) {
+            $item = array_shift($msg);
+            if ($item === '<IDS|MSG>') break;
+            else array_push($zmqIds, $item);
+        }
+
+        // Read the remaining items
+        list($hmac, $header, $parentHeader, $metadata, $content) = $msg;
 
         $header = json_decode($header, true);
         $content = json_decode($content, true);
 
         $this->logger->debug('Received message', [
             'processId' => getmypid(),
-            'zmqId' => htmlentities($zmqId, ENT_COMPAT, "UTF-8"),
-            'delim' => $delim,
+            'zmqIds' => htmlentities(implode(", ", $zmqIds), ENT_COMPAT, "UTF-8"),
             'hmac' => $hmac,
             'header' => $header,
             'parentHeader' => $parentHeader,
@@ -84,13 +92,13 @@ public function __invoke(array $msg)
         ]);
 
         if ('kernel_info_request' === $header['msg_type']) {
-            $this->kernelInfoAction->call($header, $content, $zmqId);
+            $this->kernelInfoAction->call($header, $content, $zmqIds);
         } elseif ('execute_request' === $header['msg_type']) {
-            $this->executeAction->call($header, $content, $zmqId);
+            $this->executeAction->call($header, $content, $zmqIds);
         } elseif ('history_request' === $header['msg_type']) {
-            $this->historyAction->call($header, $content, $zmqId);
+            $this->historyAction->call($header, $content, $zmqIds);
         } elseif ('shutdown_request' === $header['msg_type']) {
-            $this->shutdownAction->call($header, $content, $zmqId);
+            $this->shutdownAction->call($header, $content, $zmqIds);
         } elseif ('comm_open' === $header['msg_type']) {
             // TODO: Research about what should be done.
         } else {

src/JupyterBroker.php

@@ -51,7 +51,7 @@ public function send(
         array $content = [],
         array $parentHeader = [],
         array $metadata = [],
-        $zmqId = null
+        $zmqIds = []
     ) {
         $header = $this->createHeader($msgType);
 
@@ -62,11 +62,7 @@ public function send(
             json_encode(empty($content) ? new \stdClass : $content),
         ];
 
-        if (null !== $zmqId) {
-            $finalMsg = [$zmqId];
-        } else {
-            $finalMsg = [];
-        }
+        $finalMsg = $zmqIds;
 
         $finalMsg = array_merge(
             $finalMsg,