Megatest

(define *didsomething* #f)

;; ready? start the server
;;
(if (args:get-arg "-server")
    (let ((mode (string->symbol (args:get-arg "-server"))))
      (print "Mode: " mode)
      (case mode
	((main)(print "Starting server in main mode."))
	(else  (print "Starting server in hidden mode.")))
       ;; opens the port, drops the pkt, contacts other servers and then waits for messages
      (if (not (server:launch mode (lambda (pktrecvd)(print "Received: " pktrecvd))))
	  (exit 1))
      (set! *didsomething* #t)))

(if (args:get-arg "-repl")
    (begin
      ;; user will have to start the server manually
      (print "Run: (server:start-nmsg 'main) to start the server")
      (import extras) ;; might not be needed
      ;; (import csi)

(module
    nmsg-transport
    (
     nmsg:start-server
     nmsg:open-send-close
     nmsg:open-send-receive
     nmsg:recv
     nmsg:close
     )

(import scheme posix chicken data-structures ports)

(use nanomsg srfi-18)

       (thread-start! th1)
       (thread-start! th2)
       (thread-join! th1)
       res))))

;; default timeout is 3 seconds
;;
(define (nmsg:open-send-receive host-port msg #!key (timeout 3)(proc #f)) 
  (let ((req  (nn-socket 'req))
        (uri  (conc "tcp://" host-port))
        (res  #f))

    (handle-exceptions
     exn
     (let ((emsg ((condition-property-accessor 'exn 'message) exn)))
       ;; take action on fail
       (if proc (proc exn emsg))
       #f)
     (nn-connect req uri)

     (nn-send req msg)



     (let* ((th1  (make-thread (lambda ()
                                 (let ((resp (nn-recv req)))
                                   (nn-close req)
                                   (print resp)
                                   (set! res resp)))


                               "recv thread"))
            (th2 (make-thread (lambda ()
                                (thread-sleep! timeout)
                                (thread-terminate! th1))
                             "timer thread")))
       (thread-start! th1)
       (thread-start! th2)
       (thread-join! th1)
       res))))

(define nmsg:close nn-close)
(define nmsg:recv  nn-recv)

)

  pktsdir
  mtrah
  (mutex    (make-mutex))
  )

;; make it a global? Well, it is local to area module

(define *area-info* (make-area))
(define *pktspec*
  `((server (hostname . h)
	    (port     . p)
	    (pid      . i)
	    )
    (data   (hostname . h)  ;; sender hostname
	    (port     . p)  ;; sender port
	    (ip       . a)  ;; sender ip
	    (hostkey  . k)  ;; sending host key - store info at server under this key
	    (servkey  . s)  ;; server key - this needs to match at server end or reject the msg
	    (format   . f)  ;; sb=serialized-base64, t=text, sx=sexpr, j=json
	    (data     . d)  ;; base64 encoded slln data
	    )))

(define (server:get-mtrah)

;; look for other servers
;; contact other servers and compile list of servers
;; there are two types of server
;;     main servers - dashboards, runners and dedicated servers - need pkt
;;     passive servers - test executers, step calls, list-runs - no pkt
;;
(define (server:start-nmsg #!optional (force-server-type #f))
  (mutex-lock! (area-mutex *area-info*))
  (let* ((server-type  (or force-server-type
			   (if (args:any? "-run" "-server")
			       'main
			       'passive)))
	 (port-num     (portlogger:open-run-close portlogger:find-port))
	 (best-ip      (server:get-my-best-address))
	 (area-conn    (nmsg:start-server port-num))
	 ;; (pktspec      (area-pktspec *area-info*))
	 (mtdir        (or (server:get-mtrah)
			   (begin
			     (print "ERROR: megatest.config not found and MT_RUN_AREA_HOME is not set.")
			     #f)))
	 (pktdir       (conc mtdir
			     "/.server-pkts")))
    (if (not mtdir)
	#f
	(begin
	  (if  (not (directory? pktdir))(create-directory pktdir))
	  ;; server is started, now create pkt if needed
	  (print "Starting server in " server-type " mode")
	  (if (eq? server-type 'main)
	      (begin
		(area-pktid-set! *area-info* 
				 (write-alist->pkt
				  pktdir 
				  `((hostname . ,(get-host-name))
				    (ip       . ,best-ip)
				    (port     . ,port-num)
				    (pid      . ,(current-process-id)))
				  pktspec: *pktspec*
				  ptype:   'server))
		(area-pktfile-set! *area-info* (conc pktdir "/" (area-pktid *area-info*) ".pkt"))))
	  ;; set all the area info in the 
	  (area-pktsdir-set! *area-info* pktdir)
	  (area-mtrah-set!   *area-info* mtdir)
	  (area-conn-set!    *area-info* area-conn)
	  (area-port-set!    *area-info* port-num)
	  (mutex-unlock! (area-mutex *area-info*))
	  area-conn))))

;; Call this to start the actual server
;;
;; start_server
;;
;;   mode: '
;;   handler: proc which takes pktrecieved as argument
;;
(define (server:launch mode proc)
  (let* ((start-time (current-seconds))
	 (rep        (server:start-nmsg mode))
	 (last-msg   (current-seconds))
	 (th1        (make-thread
		      (lambda ()
			(let loop ()
			  (let ((pktdat (nmsg:recv rep)))
			    (set! last-msg (current-seconds))
			    (print "received: " pktdat)
			    (if (not (eof-object? pktdat))
				(begin
				  (proc pktdat)
				  (loop))))))
		      "message handler"))
	 (th2       (make-thread
		     (lambda ()
		       (let loop ()
			 (thread-sleep! 10)
			 (if (> (- (current-seconds) last-msg) 60) ;; timeout after 60 seconds
			     (begin
			       (print "Waited for 60 seconds and no messages, exiting now.")
			       (exit))
			     (loop)))))))
    (thread-start! th1)
    (thread-start! th2)
    (thread-join! th1)))

(define (server:shutdown)
  (let ((conn (area-conn    *area-info*))
	(pktf (area-pktfile *area-info*))
	(port (area-port    *area-info*)))
    (if conn
	(begin
	  (if pktf (delete-file* pktf))
	  (server:send-all "imshuttingdown")
	  (nmsg:close conn)
	  (portlogger:open-run-close portlogger:release-port port)))))

  (D . "1549427032.0"))

;; srvpkt is the info for the server we wish to send the message to
;;
(define (server:send servpkt data dtype)
  (let* ((port   (alist-ref 'port     servpkt))
	 (host   (alist-ref 'hostname servpkt))
	 (ip     (alist-ref 'ip       servpkt))
	 (hkey   (alist-ref 'Z        servpkt))
	 (addr   (conc (or ip host) ":" port)) ;; fall back to host if ip not provided
	 (myport (area-port *area-info*))
	 (myhost (area-myaddr *area-info*))
	 (mykey  (area-pktid  *area-info*))
	 (msg    (alist->pkt `((hostname . ,myhost)
			       (port     . ,myport)
			       (servkey  . ,hkey)     ;; server looks at this to ensure message is for them
			       (hostkey  . ,mykey)
			       (format   . ,dtype)    ;; formating of the message
			       (data     . ,data))
			     *pktspec*)))
    (if (and port host)
	(begin
	  (print "sending " msg " to " addr)
	  (nmsg:open-send-receive addr msg))
	#f)))

;; is the server alive?
;;
(define (server:ping servpkt)
  (let* ((start-time (current-milliseconds))
	 (res        (server:send servpkt "ping" "t")))
    (cons (- (current-milliseconds) start-time)
	  (equal? res "got ping"))))

;; look up all pkts and get the server id (the hash), port, host/ip
;; store this info in the global struct *area-info*
;;
(define (server:get-all)
  ;; readll all pkts
  ;; foreach pkt; if it isn't me ping the server; if alive, add to hosts hash, else rm the pkt
  (let ((all-pkts (server:get-all-server-pkts *area-info*)))
    (for-each
     (lambda (servpkt)
       (let* ((res (server:ping servpkt)))
	 (print "Got " res " from server " servpkt)))
     all-pkts)))

;; send out an "I'm about to exit notice to all known servers"
;;
(define (server:imminent-death)
  '())