Megatest

(import scheme
	chicken.base
	chicken.file
	chicken.io
	chicken.time
	chicken.condition
	chicken.port
	chicken.string
	chicken.sort
	chicken.pretty-print
	chicken.tcp
	
	address-info
	mailbox

	srfi-18
	srfi-4
	srfi-69
	system-information
	;; tcp6
	tcp-server
	typed-records

	md5
	message-digest
	(prefix base64 base64:)
	z3
	)

;; udat struct, used by both caller and callee
;; instantiated as uconn by convention
;;
(defstruct udat
  ;; the listener side

  (avail-cmboxes '())            ;; list of (<cookie> . <mbox>) for re-use
  ;; threads
  (numthreads 10)
  (cmd-thread #f)
  (work-queue-thread #f)
  (num-threads-running 0)
  ) 

;;======================================================================
;; serialization
;;   NOTE: I've had problems with read/write and s11n serialize, deserialize
;;         thus the inefficient method here
;;======================================================================

(define serializing-method (make-parameter 'complex))

;; NOTE: Can remove the regex and base64 encoding for zmq
(define (obj->string obj)
  (case (serializing-method)
    ((complex)
     (string-substitute
      (regexp "=") "_"
      (base64:base64-encode 
       (z3:encode-buffer
        (with-output-to-string
          (lambda ()(serialize obj))))) ;; BB: serialize - this is
					;; what causes problems
					;; between different builds of
					;; megatest communicating.
					;; serialize is sensitive to
					;; binary image of mtest.
      #t))
    ((write)(with-output-to-string (lambda ()(write obj))))
    ((s11n) (with-output-to-string (lambda ()(serialize obj))))
    (else obj))) ;; rpc

(define (string->obj msg #!key (transport 'http))
  (case (serializing-method)
    ((complex)
     (if (string? msg)
         (with-input-from-string 
             (z3:decode-buffer
              (base64:base64-decode
               (string-substitute 
                (regexp "_") "=" msg #t)))
           (lambda ()(deserialize)))
         (begin
           (print "ULEX ERROR: cannot translate received data \""msg"\"")
           (print-call-chain (current-error-port))
           msg))) ;; crude reply for when things go awry
    ((write)(with-input-from-string msg (lambda ()(read))))
    ((s11n)(with-input-from-string msg (lambda ()(deserialize))))
    (else msg))) ;; rpc


;;======================================================================
;; listener
;;======================================================================

;; is uconn a ulex connector (listener)
;;

(define (run-listener handler-proc #!optional (port-suggestion 4242))
  (let* ((uconn (make-udat)))
    (udat-work-proc-set! uconn handler-proc)
    (if (setup-listener uconn port-suggestion)
	((make-tcp-server
	  (udat-socket uconn)
	  (lambda ()
	    (let* ((rdat  (string->obj (read)) #;(deserialize)) ;; '(my-host-port qrykey cmd params)
		   (resp  (do-work uconn rdat)))
	      (write (obj->string resp)) #;(serialize resp)))))
	(assert #f "ERROR: run-listener called without proper setup."))))

(define (wait-and-close uconn)
  (thread-join! (udat-cmd-thread uconn))
  (tcp-close (udat-socket uconn)))

;;======================================================================

	 (isme         (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 #;(cons (current-seconds)(current-milliseconds)))))
    (cond
     (isme (do-work udata dat)) ;; no transmission needed
     (else
      (handle-exceptions ;; TODO - MAKE THIS EXCEPTION CMD SPECIFIC?
       exn
       (begin
         (print "ULEX send-receive: exn="exn)
         (message exn))
       (begin
	 ;; (mutex-lock! *send-mutex*) ;; DOESN'T SEEM TO HELP
	 (let-values (((inp oup)(tcp-connect host port)))
	   (let ((res (if (and inp oup)
			  (begin
			    (write (obj->string dat) oup) ;; (write dat oup);; (serialize dat oup)
			    (close-output-port oup)
			    (string->obj (read inp))) ;; (deserialize inp))
			  (begin
			    (print "ERROR: send called but no receiver has been setup. Please call setup first!")
			    #f))))
	     ;; (close-output-port oup)
	     (close-input-port inp)
	     ;; (mutex-unlock! *send-mutex*) ;; DOESN'T SEEM TO HELP
	     res)))))))) ;; res will always be 'ack unless return-method is direct

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

;; move the logic to return the result somewhere else?
;;