;;;   TODO:
;;;   =====
;;;   
;;;    o  Zyklenerkennung?
;;;    o  nicht benoetigte Threads runterfahren

(define-record-type :rule
  (really-make-rule prereqs wants-build? build-func)
  is-rule?
  (prereqs rule-prereqs)
  (wants-build? rule-wants-build?)
  (build-func rule-build-func))

(define-enumerated-type rule-cmd :rule-cmd
  is-rule-cmd?
  the-rule-cmds
  rule-cmd-name
  rule-cmd-index
  (make link shutdown))

(define (rule-make rule init-state)
  (let* ((server (let ((found? (assq rule rules)))
		   (if (is-collect&reply-channel? (cdr found?)) 
		       (cdr found?) 
		       (error "rule-make: rule not found."))))
	 (client (send&collect/make-channel))
	 (recipient (make-link client server)))
    (send&collect/send client (make-tagged-msg recipient (rule-cmd make)))
    (send&collect/send client (make-tagged-msg recipient init-state))
    (tagged-msg-stripped (send&collect/receive client))))

(define rules (list))
(define lock-rules (make-lock))

(define (make-rule prereqs wants-build? build-func)
  (let ((rule (really-make-rule prereqs wants-build? build-func))
	(listen-ch (collect&reply/make-channel)))
    (with-lock lock-rules 
      (lambda () 
	(if (not (assq rule rules))
	    (begin 
	      (set! rules (alist-cons rule listen-ch rules))
	      (rule-node rule listen-ch))
	    (error "make-rule: rule already exists."))))
    rule))

(define (rule-node/sort-msgs unsorted to-order)
  (map (lambda (pos)
	 (map (lambda (tmsg)
		(let ((msg (tagged-msg-stripped tmsg))
		      (sender (tagged-msg-tag tmsg)))
		  (if (eq? sender pos)
		      msg)))
	      unsorted))
       to-order))

(define (rule-node/make rule recipients connect-ch listen-ch init-state)
  (let* ((to-sort (map (lambda (recipient)
			 (let ((tmsg-cmd (make-tagged-msg recipient 
							  (rule-cmd make)))
			       (tmsg-state (make-tagged-msg recipient 
							    init-state)))
			   (send&collect/send connect-ch tmsg-cmd)
			   (send&collect/send connect-ch tmsg-state)
			   (send&collect/receive connect-ch)))
		       recipients))
	 (res-pres (rule-node/sort-msgs to-sort recipients))
	 (res-build? (call-with-values
			 (lambda ()
			   (apply values (append res-pres 
						 (list init-state))))
		       (rule-wants-build? rule)))
	 (res-wants-build? (car res-build?))
	 (cooked-state (cdr res-build?)))
    (if res-wants-build? 
	(let ((build-res (call-with-values 
			     (lambda () 
			       (apply values (append (list res-wants-build?)
						     res-pres
						     (list cooked-state))))
			   (rule-build-func rule))))
	  build-res)
	(cons #t cooked-state))))

(define (rule-node/recipients rule connect-ch)
  (let ((server-chs (map (lambda (r) 
			   (with-lock lock-rules
			     (lambda () 
			       (cdr (assq r rules)))))
			 (rule-prereqs rule))))
    (map (lambda (server-ch)
	   (make-link connect-ch server-ch))
	 server-chs)))

(define (rule-node rule listen-ch)
  (let ((connect-ch (send&collect/make-channel)))
    (spawn
      (lambda () 
	(let node-loop ((tmsg (collect&reply/receive listen-ch))
			(recipients #f))
	  (let ((sender (tagged-msg-tag tmsg))
		(cmd (tagged-msg-stripped tmsg)))
	    (cond
	     ((eq? (rule-cmd-name cmd) 'make)
	      (if (not recipients) 
		  (set! recipients (rule-node/recipients rule connect-ch)))
	      (let* ((tmsg (collect&reply/receive listen-ch))
		     (init-state (tagged-msg-stripped tmsg))
		     (res (rule-node/make rule recipients
					    connect-ch listen-ch init-state)))
		(collect&reply/send listen-ch (make-tagged-msg sender res))))
	     ((eq? (rule-cmd-name cmd) 'shutdown)
	      (terminate-current-thread))))
	  (node-loop (collect&reply/receive listen-ch) recipients)))
      'rule-node)))