mq-impl/stream: Allow closing the queue on request.

* gnu/gnunet/mq.scm
  (make-message-queue): Accept optional 'closer' argument.
  (close-queue!): New procedure.
* gnu/gnunet/mq-impl/stream.scm
  (prepare-port-message-queue): Remove fixed TODO.  Also return
  'close!'.
  (prepare-port-message-queue)[request-close-condition]: New variable.
  (prepare-port-message-queue)[start-reader!]{wait-op}: Wait on
  request-close-condition and escape.
  (prepare-port-message-queue)[close!]: New procedure.
  (port->message-queue): Accept 'close!' from
  'prepare-port-message-queue'.  Pass it to the message queue
  constructor.
  (connect/fibers): Accept 'close!' from
  'prepare-port-message-queue'.  Pass it to 'make-message-queue'.
  Note a TODO.
* tests/mq-stream.scm
  ("fibers stop and port closed after close! (directly after creation)")
  ("fibers stop and port closed after close! (some times passes)"):
  New tests.

1 files changed, 22 insertions, 0 deletions

diff --git a/tests/mq-stream.scm b/tests/mq-stream.scm
index e16aa1b..f5a56e6 100644
--- a/tests/mq-stream.scm
+++ b/tests/mq-stream.scm
@@ -729,4 +729,26 @@ If an EPIPE system error is raised, return #f."
      (port-closed? alpha))
    #:parallelism 1)) ; to make the use of yield-many less fragile
 
+(test-assert "fibers stop and port closed after close! (directly after creation)"
+  (let^ ((<-- (alpha beta) (two-sockets)))
+        (call-with-spawner/wait
+         (lambda (spawn)
+           (define q (port->message-queue alpha no-handlers error-handler/regular
+                                          #:spawn spawn))
+           (close-queue! q))
+         #:parallelism 1)
+        (port-closed? alpha)))
+
+(test-assert "fibers stop and port closed after close! (some times passes)"
+  (let^ ((<-- (alpha beta) (two-sockets)))
+        (call-with-spawner/wait
+         (lambda (spawn)
+           (define q (port->message-queue alpha no-handlers error-handler/regular
+                                          #:spawn spawn))
+           (yield-many)
+           (sleep 0.01)
+           (close-queue! q))
+         #:parallelism 1)
+        (port-closed? alpha)))
+
 (test-end "mq-stream")