Megatest

(import scheme
	chicken.base
	chicken.file
	chicken.time
	chicken.condition
	chicken.string
	chicken.sort
	chicken.pretty-print
	
	address-info
	mailbox
	matchable
	queues
	regex
	regex-case

;;
;; NOTE: see below for beginnings of code to allow re-use of tcp connections
;;        - I believe (without substantial evidence) that re-using connections will
;;          be beneficial ...
;;
(define (send udata host-port qrykey cmd params)
  (let* ((my-host-port (udat-host-port udata))          ;; remote will return to this
	 (isme         (equal? host-port my-host-port)) ;; calling myself?
	 (isme         #f #;(equal? host-port my-host-port)) ;; calling myself?
	 ;; dat is a self-contained work block that can be sent or handled locally
	 (dat          (list my-host-port qrykey cmd params))
	 )
    (if isme
	(ulex-handler udata dat) ;; no transmission needed
	(handle-exceptions ;; TODO - MAKE THIS EXCEPTION CMD SPECIFIC?
	    exn

	 (mbox      (cdr cmbox))
	 (mbox-time (current-milliseconds)))
    (if (eq? (send uconn host-port qrykey cmd data) 'ack)
	(let* ((mbox-timeout-secs    120) ;; timeout)
	       (mbox-timeout-result 'MBOX_TIMEOUT)
	       (res                  (mailbox-receive! mbox mbox-timeout-secs mbox-timeout-result))
	       (mbox-receive-time    (current-milliseconds)))
	  (print "In send-receive, got "res" back from mailbox")
	  (if (eq? res 'MBOX_TIMEOUT)
	      #f  ;; convert to raising exception?
	      res))
	#f))) ;; #f means failed to communicate

;;======================================================================
;; responder side
;;======================================================================

;; take a request, rdata, and if not immediate put it in the work queue
;;
;; Reserved cmds; ack ping goodbye response
;;
(define (ulex-handler uconn rdata)
  (print "ulex-handler received data: "rdata)
  (match rdata ;;  (string-split controldat)
    ((rem-host-port qrykey cmd params) ;; cmdkey host-port pid qrykey params ...)
     (let ((mbox (hash-table-ref/default (udat-mboxes uconn) qrykey #f)))
     (case cmd
       ((ack )(print "Got ack! But why? Should NOT get here.") 'ack)
       ((ping) 'ack) ;; special case - return result immediately on the same connection
       ((goodbye)
	;; just clear out references to the caller
	'ack)
       ((response) ;; this is a result from remote processing, send it as mail ...
       (case cmd
	 ;; ((ack )(print "Got ack! But why? Should NOT get here.") 'ack)
	 ((ping)
	  (print "Got Ping!")
	  (add-to-work-queue uconn rdata)
	 'ack)
	 ((goodbye)
	  ;; just clear out references to the caller
	  (add-to-work-queue uconn rdata)
	  'ack)
	 ((response) ;; this is a result from remote processing, send it as mail ...
	(let ((mbox (hash-table-ref/default (udat-mboxes uconn) qrykey #f)))
	  (if mbox
	      (begin
	      (mailbox-send! mbox params) ;; params here is our result
		(mailbox-send! mbox params) ;; params here is our result
		'ack)
	      (begin
		(print "ERROR: received result but no associated mbox for cookie "qrykey)
		#f))))
       ((else
	(add-to-work-queue uconn rdata)
	'ack))))
		#f)))
	 ((else
	   (add-to-work-queue uconn rdata)
	   'ack)))))
    (else
     (print "BAD DATA? controldat=" rdata)
     'ack) ;; send ack anyway?
    ))

;; given an already set up uconn start the cmd-loop
;;

  (udat-work-proc-set! uconn proc))

;; run-listener does all the work of starting a listener in a thread
;; it then returns control
;;
(define (run-listener handler-proc)
  (let* ((uconn (make-udat)))
    (udat-work-proc-set! uconn handler-proc)
    (if (setup-listener uconn)
	(let* ((th1 (make-thread (lambda ()(ulex-cmd-loop uconn)) "Ulex command loop"))
	       (th2 (make-thread (lambda ()(process-work-queue uconn)) "Ulex work queue processor")))
	  (thread-start! th1)
	  (thread-start! th2)
	  )
	  (print "cmd loop and process workers started")
	  uconn)
	(begin
	  (print "ERROR: run-listener called without proper setup.")
	  (exit)))))

;;======================================================================
;; work queues - this is all happening on the listener side
;;======================================================================

;; rdata is (rem-host-port qrykey cmd params)
					     
(define (add-to-work-queue uconn rdata)
  (queue-add! (udat-work-queue uconn) rdata))

(define (do-work uconn rdata)
  (let* ((proc (udat-work-proc uconn))) ;; get it each time - conceivebly it could change
    ;; put this following into a do-work procedure
    (match rdata
      ((rem-host-port qrykey cmd params)
       (let* ((result (proc rem-host-port qrykey cmd params)))
	 ;; send 'response as cmd and result as params
	 (send uconn rem-host-port qrykey result))) ;; could check for ack
	 (send uconn rem-host-port qrykey 'response result))) ;; could check for ack
      (else
       (print "ERROR: rdata "rdata", did not match rem-host-port qrykey cmd params")))))
     
     
(define (process-work-queue uconn) 
  (let ((wqueue (udat-work-queue uconn))
	(proc   (udat-work-proc  uconn)))

;; cookie/mboxes

;; we store each mbox with a cookie (<cookie> . <mbox>)
;;
(define (get-cmbox uconn)
  (if (null? (udat-avail-cmboxes uconn))
      (let ((cookie (make-cookie))
      (let ((cookie (make-cookie uconn))
	    (mbox   (make-mailbox)))
	(hash-table-set! (udat-mboxes uconn) cookie mbox)
	`(cookie . mbox))
	`(,cookie . ,mbox))
      (let ((cmbox (car (udat-avail-cmboxes uconn))))
	(udat-avail-cmboxes-set! uconn (cdr (udat-avail-cmboxes uconn)))
	cmbox)))

(define (put-cmbox uconn cmbox)
  (udat-avail-cmboxes-set! uconn (cons cmbox (udat-avail-cmboxes uconn))))

;; peers
(define (pp-uconn uconn)
  (pp (udat->alist uconn)))

  
;;======================================================================
;; network utilities
;;======================================================================

;; NOTE: Look at address-info egg as alternative to some of this

       (filter (lambda (x)
		 (equal? (address-info-type x) "tcp"))
	       (address-infos (get-host-name)))))

;; (map ip->string (vector->list 
;; 		   (hostinfo-addresses
;; 		    (host-information (current-hostname))))))

)

(import ulex)

)

(import ulex trace big-chicken srfi-18)
(trace-call-sites #t)
(trace ulex-handler
       send)
 
(define (handler-proc . data)
  (print "handler-proc, got: "data)
  `(data ,data))

(define uconn (run-listener handler-proc))

(pp-uconn uconn)

(define res #f)
(define th1 (make-thread (lambda ()
			   (set! res (send-receive uconn "zeus:4242" 'ping '())))))
(thread-start! th1)
(thread-join! th1)
		
(thread-sleep! 1)
(print "All done")
(print "Received "res)