8sync/inbox.scm

   1 ;;; 8sync --- Asynchronous programming for Guile
   2 ;;; Copyright © 2017 Christopher Allan Webber <cwebber@dustycloud.org>
   3 ;;;
   4 ;;; This file is part of 8sync.
   5 ;;;
   6 ;;; 8sync is free software: you can redistribute it and/or modify it
   7 ;;; under the terms of the GNU Lesser General Public License as
   8 ;;; published by the Free Software Foundation, either version 3 of the
   9 ;;; License, or (at your option) any later version.
  10 ;;;
  11 ;;; 8sync is distributed in the hope that it will be useful,
  12 ;;; but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 ;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 ;;; GNU Lesser General Public License for more details.
  15 ;;;
  16 ;;; You should have received a copy of the GNU Lesser General Public
  17 ;;; License along with 8sync.  If not, see <http://www.gnu.org/licenses/>.
  18
  19 (define-module (8sync inbox)
  20   #:use-module (fibers)
  21   #:use-module (fibers channels)
  22   #:use-module (fibers conditions)
  23   #:use-module (fibers operations)
  24   #:use-module (ice-9 match)
  25   #:use-module (ice-9 q)
  26   #:use-module (srfi srfi-9)
  27   #:use-module (ice-9 atomic)
  28   #:export (delivery-agent))
  29
  30 (define (delivery-agent inbox-enq inbox-deq stop?)
  31   "This starts up a loop doing delivery receiving from INBOX-ENQ and
  32 delivering to INBOX-DEQ, actually managing an (ice-9 q) object QUEUE.
  33 Atomic box STOP? can be set to indicate that this "
  34   (define queue
  35     (make-q))
  36   (define get-or-stop
  37     (choice-operation
  38      (wrap-operation (get-operation inbox-enq)
  39                      (lambda (message)
  40                        (enq! queue message)
  41                        'got-one))
  42      (wrap-operation (wait-operation stop?)
  43                      (const 'stop))))
  44   (let main-lp ()
  45     (cond
  46      ;; No items to deliver?  We need to get one first...
  47      ((q-empty? queue)
  48       (match (perform-operation get-or-stop)
  49         ;; keep looping
  50         ('got-one (main-lp))
  51         ;; halt!
  52         ('stop 'done)))
  53      (else
  54       ;; Pull an item off the queue for delivery...
  55       (let ((this-one (deq! queue)))
  56         ;; But we need to start looping!
  57         (let deliver-this-one ()
  58           (match (perform-operation
  59                   (choice-operation
  60                    ;; get a new message and keep trying to deliver
  61                    ;; this one, or stop
  62                    get-or-stop
  63                    ;; deliver this one and get the next one to deliver
  64                    (wrap-operation (put-operation inbox-deq this-one)
  65                                    (const 'delivered))))
  66             ;; We're dispatching based on which one succeeds.
  67             ;; Maybe this isn't necessary, but I'm not convinced
  68             ;; that looping within the choice-operation would be
  69             ;; properly tail recursive.
  70             ('got-one (deliver-this-one))
  71             ('delivered (main-lp))
  72             ('stop 'done))))))))