messaging: Make "execute_request" more robust.

* tests/kernels.scm ("execute_request"): Rewrite to remove assumptions
about ordering, and to account for all the replies.

tests/kernels.scm

@@ -21,7 +21,9 @@
   #:use-module (json)
   #:use-module (rnrs bytevectors)
   #:use-module (srfi srfi-1)
-  #:use-module (srfi srfi-64))
+  #:use-module (srfi srfi-26)
+  #:use-module (srfi srfi-64)
+  #:use-module (ice-9 match))
 
 (define %context
   (zmq-create-context))
@@ -97,28 +99,49 @@
     ;; in the request-reply pattern, we first receive on the iopub socket a
     ;; "status" message, then our input is broadcast on iopub, and then we
     ;; get the result.
-    (let* ((reply1  (read-message %kernel))
-           (reply2  (read-message %kernel))
-           (reply3  (read-message %kernel))
-           (replies (pk 'replies (list reply1 reply2 reply3))))
-      (and (every message? replies)
-           (every (compose bytevector? message-sender) replies)
-
-           (string=? "status" (message-type reply1))
-           (string=? "busy"
-                     (assoc-ref (json-string->scm (message-content reply1))
-                                "execution_state"))
-           (equal? (message-parent-header reply1) (message-header request))
-
-           (string=? "execute_input" (message-type reply2))
-           (equal? (message-parent-header reply2) (message-header request))
-
-           (string=? (message-type reply3) "execute_result")
-           (equal? (message-parent-header reply3) (message-header request))
-           (let* ((content (json-string->scm (message-content reply3)))
-                  (data    (assoc-ref content "data"))
-                  (text    (assoc-ref data "text/plain")))
-             (string->number text))))))
+    ;;
+    ;; However, ordering is not guaranteed: since we poll on all these
+    ;; sockets, we might get the iopub message before the shell message.
+    ;; Thus, this test does not make any assumptions on ordering.
+    (let* ((replies (unfold (cut > <> 4)
+                            (lambda (_)
+                              (read-message %kernel 10000))
+                            1+ 0)))
+      (define (type-predicate type)
+        (lambda (message)
+          (string=? (message-type message) type)))
+
+      (and (every message? (pk 'replies replies))
+
+           (match (filter (type-predicate "status") replies)
+             ((busy idle)
+              ;; We got two message: busy *then* idle.
+              (and (string=? "busy"
+                             (assoc-ref (json-string->scm
+                                         (message-content busy))
+                                        "execution_state"))
+                   (string=? "idle"
+                             (assoc-ref (json-string->scm
+                                         (message-content idle))
+                                        "execution_state"))
+                   (equal? (message-parent-header busy)
+                           (message-header request))
+                   (equal? (message-parent-header idle)
+                           (message-header request)))))
+
+           (let ((input (find (type-predicate "execute_input") replies)))
+             (equal? (message-parent-header input) (message-header request)))
+
+           (let ((reply (find (type-predicate "execute_reply") replies)))
+             (equal? (message-parent-header reply) (message-header request)))
+
+           (let ((result (find (type-predicate "execute_result") replies)))
+             (and (equal? (message-parent-header result)
+                          (message-header request))
+                  (let* ((content (json-string->scm (message-content result)))
+                         (data    (assoc-ref content "data"))
+                         (text    (assoc-ref data "text/plain")))
+                    (string->number text))))))))
 
 (when %kernel
   (close-kernel %kernel)