;; Einfache RPC-Protokolle

(define-record-type :rpc-op
  (make-rpc-op call entry-proc)
  rpc-op?
  (call rpc-op-call)
  (entry-proc rpc-op-entry-proc))

(define (make-rpc server-op)
  (let ((request-channel (make-channel)))
    (make-rpc-op
     (lambda (arg)
       (let ((reply-placeholder (make-placeholder)))
	 (send request-channel
	       (cons arg reply-placeholder))
	 (placeholder-value reply-placeholder)))
     (lambda (state)
       (wrap
	(receive-rv request-channel)
	(lambda (pair)
	  (let ((arg (car pair))
		(reply-placeholder (cdr pair)))
	    (call-with-values
	     (lambda () (server-op state arg))
	     (lambda (new-state result)
	       (placeholder-set! reply-placeholder result)
	       new-state)))))))))

(define (make-rpc-server initial-state entries)
  (spawn
   (lambda ()
     (let loop ((state initial-state))
       (loop
	(select (map (lambda (entry-proc)
		       (entry-proc state))
		     entries)))))))

(define (make-conditional-rpc pred server-op)
  (let ((request-channel (make-channel))
	(pred-channel (make-channel))
	(forward-channel (make-channel)))

    (condition-buffer (receive-rv pred-channel)
		      (receive-rv request-channel)
		      forward-channel)

    (make-rpc-op
     (lambda (arg)
       (let ((reply-placeholder (make-placeholder)))
	 (send request-channel
	       (cons arg reply-placeholder))
	 (placeholder-value reply-placeholder)))
     (lambda (state)
       (with-nack
	(lambda (nack-rv)
	  (send pred-channel (cons (lambda (arg) (pred state arg))
				   nack-rv))
	  (wrap
	   (receive-rv forward-channel)
	   (lambda (pair)
	     (let ((arg (car pair))
		   (reply-placeholder (cdr pair)))
	       (call-with-values
		(lambda () (server-op state arg))
		(lambda (new-state result)
		  (placeholder-set! reply-placeholder result)
		  new-state)))))))))))

(define (condition-buffer pred-rv
			  request-rv
			  forward-channel)

  (define (forward-call call disable)
    (select
     (list
      (wrap disable
	    (lambda (ignore)
	      #f))
      (wrap (send-rv forward-channel call)
	    (lambda (ignore)
	      #t)))))

  (define (scan-conditions pred calls)
    (let loop ((calls calls)
	       (deferred '()))
      (cond
       ((null? calls) #f)
       ((pred (car (car calls)))
	(cons (car calls)
	      (append (reverse deferred)
		      (cdr calls))))
       (else
	(loop (cdr calls) (cons (car calls) deferred))))))

  (define (enabled calls pred disable)
    (select
     (list
      (wrap disable
	    (lambda (ignore) (disabled calls)))
      (wrap request-rv
	    (lambda (call)
	      (let ((arg (car call)))
		(cond
		 ((not (pred arg))
		  (enabled (append calls (list call))
			   pred
			   disable))
		 ((forward-call call disable)
		  (disabled calls))
		 (else
		  (disabled (append calls (list call)))))))))))

  (define (disabled calls)
    (select
     (list
      (wrap request-rv
	    (lambda (call)
	      (disabled (append calls (list call)))))
      (wrap pred-rv
	    (lambda (pair)
	      (let ((pred (car pair))
		    (disable (cdr pair)))
		(cond
		 ((scan-conditions pred calls)
		  => (lambda (pair)
		       (let ((call (car pair))
			     (new-calls (cdr pair)))
			 (if (forward-call call disable)
			     (disabled new-calls)
			     (disabled calls)))))
		 (else
		  (enabled calls pred disable)))))))))

  (spawn
   (lambda ()
     (disabled '()))))