Added zmq identifier to action calls

Author: bstoney

src/Actions/Action.php

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

src/Actions/ExecuteAction.php

@@ -52,7 +52,7 @@ public function __construct(
         $this->shellSoul = $shellSoul;
     }
 
-    public function call(array $header, array $content)
+    public function call(array $header, array $content, $zmqId = null)
     {
         $this->broker->send(
             $this->iopubSocket, 'status', ['execution_state' => 'busy'], $header

src/Actions/HistoryAction.php

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

src/Actions/KernelInfoAction.php

@@ -32,7 +32,7 @@ public function __construct(JupyterBroker $broker, SocketWrapper $shellSocket, S
         $this->iopubSocket = $iopubSocket;
     }
 
-    public function call(array $header, array $content)
+    public function call(array $header, array $content, $zmqId = null)
     {
         // TODO: Implement call() method.
 

src/Actions/ShutdownAction.php

@@ -6,7 +6,7 @@
 
 final class ShutdownAction implements Action
 {
-    public function call(array $header, array $content)
+    public function call(array $header, array $content, $zmqId = null)
     {
         // TODO: Implement call() method.
     }

src/Handlers/ShellMessagesHandler.php

@@ -75,7 +75,7 @@ public function __invoke(array $msg)
 
         $this->logger->debug('Received message', [
             'processId'    => getmypid(),
-            'zmqId'        => $zmqId,
+            'zmqId'        => htmlentities($zmqId, ENT_COMPAT, "UTF-8"),
             'delim'        => $delim,
             'hmac'         => $hmac,
             'header'       => $header,
@@ -85,13 +85,13 @@ public function __invoke(array $msg)
         ]);
 
         if ('kernel_info_request' === $header['msg_type']) {
-            $this->kernelInfoAction->call($header, $content);
+            $this->kernelInfoAction->call($header, $content, $zmqId);
         } elseif ('execute_request' === $header['msg_type']) {
-            $this->executeAction->call($header, $content);
+            $this->executeAction->call($header, $content, $zmqId);
         } elseif ('history_request' === $header['msg_type']) {
-            $this->historyAction->call($header, $content);
+            $this->historyAction->call($header, $content, $zmqId);
         } elseif ('shutdown_request' === $header['msg_type']) {
-            $this->shutdownAction->call($header, $content);
+            $this->shutdownAction->call($header, $content, $zmqId);
         } elseif ('comm_open' === $header['msg_type']) {
             // TODO: Research about what should be done.
         } else {

src/JupyterBroker.php

@@ -57,11 +57,16 @@ public function __construct($key, $signatureScheme, UuidInterface $sessionId, Lo
      * @param array $content
      * @param array $parentHeader
      * @param array $metadata
+     * @param string $zmqId
      */
     public function send(
-        SocketWrapper $stream, $msgType, array $content = [], array $parentHeader = [], array $metadata = []
-    )
-    {
+        SocketWrapper $stream,
+        $msgType,
+        array $content = [],
+        array $parentHeader = [],
+        array $metadata = [],
+        $zmqId = null
+    ) {
         $header = $this->createHeader($msgType);
 
         $msgDef = [
@@ -71,10 +76,16 @@ public function send(
             json_encode(empty($content) ? new \stdClass : $content),
         ];
 
+        if ($zmqId !== null) {
+            $finalMsg = [$zmqId];
+        } else {
+            $finalMsg = [];
+        }
+
         $finalMsg = array_merge(
+            $finalMsg,
             ['<IDS|MSG>', $this->sign($msgDef)],
-            $msgDef
-        );
+            $msgDef);
 
         if (null !== $this->logger) {
             $this->logger->debug('Sent message', ['processId' => getmypid(), 'message' => $finalMsg]);
@@ -90,15 +101,17 @@ public function send(
     private function createHeader($msgType)
     {
         return [
-            'date'     => (new \DateTime('NOW'))->format('c'),
-            'msg_id'   => Uuid::uuid4()->toString(),
+            'date' => (new \DateTime('NOW'))->format('c'),
+            'msg_id' => Uuid::uuid4()->toString(),
             'username' => "kernel",
-            'session'  => $this->sesssionId->toString(),
+            'session' => $this->sesssionId->toString(),
             'msg_type' => $msgType,
+            'version' => '5.0',
         ];
     }
 
-    private function sign(array $message_list) {
+    private function sign(array $message_list)
+    {
         $hm = hash_init(
             $this->hashAlgorithm,
             HASH_HMAC,