Megatest

(declare (uses itemsmod))
(declare (uses mtargs))
(declare (uses mtver))
(declare (uses pgdb))
(declare (uses portloggermod))
(declare (uses servermod))
(declare (uses tasksmod))


(module rmtmod
	*
	
(import scheme
		
	chicken.base

	directory-utils
	format
	;; http-client
	;; intarweb
	matchable
	md5
	message-digest
	nng ;; nanomsg
	(prefix base64 base64:)
	(prefix sqlite3 sqlite3:)
	regex
	s11n
	;; spiffy
	;; spiffy-directory-listing
	;; spiffy-request-vars

	pgdb
	pkts
	portloggermod
	(prefix mtargs args:)
	servermod
	stml2
	tasksmod


	)

(defstruct alldat
  (areapath #f)
  (ulexdat  #f)
  )

  (port #f)
  (uuid #f)
  (rep  #f)
  (dbfile #f)
  (api-url #f)
  (api-uri #f)
  (api-req #f)


  (status 'starting)
  (trynum 0) ;; count the number of ports we've tried
  ) 

(define (servdat->url sdat)
  (conc (servdat-host sdat)":"(servdat-port sdat)))

;; -> http://abc.com:900/<entrypoint>
;;
(define (conndat->uri conn entrypoint)
  (conc "http://"(conndat-ipaddr conn)":"(conndat-port conn)"/"entrypoint))

;; set up the api proc, seems like there should be a better place for this?



(define api-proc (make-parameter conc))
(api-proc api:process-request)

;; do we have a connection to apath dbname and
;; is it not expired? then return it
;;
;; else setup a connection
;;
;; if that fails, return '(#f "some reason") ;; NB// convert to raising an exception
;;
(define (rmt:get-conn remdat apath dbname)
  (let* ((fullname (db:dbname->path apath dbname))
	 (conn     (hash-table-ref/default (remotedat-conns remdat) fullname #f)))
    (if (and conn
	     (< (current-seconds) (conndat-expires conn)))
	conn
	#f)))

(define (rmt:find-main-server apath dbname)
  (let* ((pktsdir     (get-pkts-dir apath))
	 (all-srvpkts (get-all-server-pkts pktsdir *srvpktspec*))
	 ;; (dbpath      (conc apath "/" dbname))
	 (viable-srvs (get-viable-servers all-srvpkts dbname)))
    (get-the-server apath viable-srvs)))


(define *connstart-mutex* (make-mutex))
(define *last-main-start* 0)

;; looks for a connection to main, returns if have and not exired
;; creates new otherwise
;; 
;; connections for other servers happens by requesting from main
;;
;; TODO: This is unnecessarily re-creating the record in the hash table
;;
(define (rmt:open-main-connection remdat apath)
  (let* ((fullpath (db:dbname->path apath "/.db/main.db"))
	 (conns    (remotedat-conns remdat))
	 (conn     (hash-table-ref/default conns fullpath #f))) ;; TODO - create call for this













    (cond
     ((and conn                                             ;; conn is NOT a socket, just saying ...
	   (< (current-seconds) (conndat-expires conn)))
      #t) ;; we are current and good to go - we'll deal elsewhere with a server that was killed or died 
     ((and conn
	   (>= (current-seconds)(conndat-expires conn)))
      (debug:print-info 0 *default-log-port* "connection to "fullpath" server expired. Reconnecting.")
      (if (conndat-socket conn)
	  (nng-close! (conndat-socket conn)))
      (hash-table-set! conns fullpath #f) ;; clean up
      (rmt:open-main-connection remdat apath))
     (else
      ;; Below we will find or create and connect to main
      (let* ((dbname         (db:run-id->dbname #f))
	     (the-srv        (rmt:find-main-server apath dbname))
	     (start-main-srv (lambda () ;; call IF there is no the-srv found
			       (mutex-lock! *connstart-mutex*)
			       (if (> (- (current-seconds) *last-main-start*) 5) ;; at least four seconds since last attempt to start main server
				   (begin
				     (api:run-server-process apath dbname)
				     (set! *last-main-start* (current-seconds))
				     (thread-sleep! 1)))

		   (fullpath (db:dbname->path apath dbname))
		   
		   (new-the-srv (make-conndat
				 apath:   apath
				 dbname:  dbname
				 fullname: fullpath
				 hostport: srv-addr
				 socket: (open-nn-connection srv-addr)
				 ipaddr: ipaddr
				 port: port
				 srvpkt: the-srv
				 srvkey: srvkey ;; generated by rmt:get-signature on the server side
				 lastmsg: (current-seconds)
				 expires: (+ (current-seconds) 60) ;; this needs to be gathered during the ping
				 )))

	  (debug:print-logger rmt:log-to-main)))
    (cond
     ((or (not mconn) ;; no channel open to main?
	  (< (conndat-expires mconn)(+ (current-seconds) 2))) ;; restablish connection if less than 2 seconds on the lease
      (if mconn ;; previously opened - clean up NB// consolidate this with the similar code in open main above
	  (begin
	    (debug:print-info 0 *default-log-port* "Clearing out connection to main that has expired.")
	    (nng-close! (conndat-socket mconn))
	    (hash-table-set! conns fullname #f)))
      (rmt:open-main-connection remdat apath)
      (rmt:general-open-connection remdat apath mdbname))
     ((not (rmt:get-conn remdat apath dbname))                 ;; no channel open to dbname?     
      (let* ((res (rmt:send-receive-real remdat apath mdbname 'get-server `(,apath ,dbname))))
	(case res
	  ((server-started)

		   (debug:print-info 0 *default-log-port* "got "res)
		   (hash-table-set! conns
				    fullname
				    (make-conndat
				     apath: apath
				     dbname: dbname
				     hostport: (conc host":"port)
				     socket: (open-nn-connection (conc host":"port))
				     ipaddr: ipaddr
				     port: port
				     srvkey: servkey
				     lastmsg: (current-seconds)
				     expires: (+ (current-seconds) 60))))
		  (else
		   (debug:print-info 0 *default-log-port* "return data from starting server did not match host port servkey pid ipaddr apath dbname " res)))

  (let* ((apath      *toppath*)
	 (remdat     *remotedat*)
	 (conns      (remotedat-conns remdat)) ;; just checking that remdat is a remotedat
	 (dbname     (db:run-id->dbname rid)))
    (if *localmode*
	(let* ((dbdat    (dbr:dbstruct-get-dbdat *dbstruct* dbname))
	       (indat    `((cmd . ,cmd)(params . ,params))))

	  (api:process-request *dbstruct* indat)
	  ;; (api:process-request dbdat indat)
	  )
	(begin
	  (rmt:open-main-connection remdat apath)
	  (if rid (rmt:general-open-connection remdat apath dbname))
	  (rmt:send-receive-real remdat apath dbname cmd params)))))

    (let* ((soc     (conndat-socket conn))
	   (key     #f)
	   (host    (conndat-ipaddr conn))
	   (port    (conndat-port   conn))
	   (payload `((cmd    . ,cmd)
		      (key    . ,(conndat-srvkey conn))
		      (params . ,params)))
	   (res      (send-receive-nn soc ;; (open-send-receive-nn (conc host":"port)
				      (sexpr->string payload))))
      (if (member res '("#<unspecified>")) ;; TODO - fix this in string->sexpr
	  #f
	  (string->sexpr res)))))

;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future.
;;

	(let* ((am-server  (args:get-arg "-server"))
	       (dbfile     (args:get-arg "-db"))
	       (apath      *toppath*)
	       (remdat     *remotedat*)) ;; foundation for future fix
	  (if *dbstruct-db*
	      (let* ((dbdat      (db:get-dbdat *dbstruct-db* apath dbfile))
		     (db         (dbr:dbdat-db dbdat))
		     (inmem      (dbr:dbdat-db dbdat))
		     )
		;; do a final sync here
		(debug:print-info 0 *default-log-port* "Doing final sync for "apath" "dbfile" at "(current-seconds))
		(db:sync-inmem->disk *dbstruct-db* apath dbfile force-sync: #t)
		;; let's finalize here
		(debug:print-info 0 *default-log-port* "Finalizing db and inmem")
		(if (sqlite3:database? db)

			 #t))))
    (debug:print-info 4 *default-log-port* "starting exit process, finalizing databases.")
    (if (and no-hurry (debug:debug-mode 18))
	(rmt:print-db-stats))
    (let ((th1 (make-thread
		(lambda () ;; thread for cleaning up, give it five seconds
		  (let* ((start-time (current-seconds)))
		    (if (and *server-info*
			     *unclean-shutdown*)
			(begin
			  (debug:print-info 0 *default-log-port* "Unclean server exit, calling server-shtudown")



			  (rmt:server-shutdown (servdat-host *server-info*)

					       (servdat-port *server-info*))))
		    (debug:print-info 0 *default-log-port* "Shutdown activities completed in "(- (current-seconds) start-time)" seconds"))
		  ;; (if *dbstruct-db* (db:close-all *dbstruct-db*)) ;; one second allocated
		  #;(if (bdat-task-db *bdat*)    ;; TODO: Check that this is correct for task db
		  (let ((db (cdr (bdat-task-db *bdat*))))
		  (if (sqlite3:database? db)
		  (begin
		  (debug:print-info 0 *default-log-port* "Closing down task db "db)

;;
;; conn is a conndat record
;;
(define (server:ping conn #!key (do-exit #f))
  (let* ((req (conndat-socket conn))
	 (srvkey (conndat-srvkey conn))
	 (msg (sexpr->string '(ping ,srvkey))))
    (send-receive-nn req msg))) ;; (server-ready? host port server-id))

;;======================================================================
;; http-transportmod.scm contents moved here
;;======================================================================

(define (http-transport:make-server-url hostport)
  (if (not hostport)
      #f
      (conc "http://" (car hostport) ":" (cadr hostport))))

;;======================================================================
;; S E R V E R
;; ======================================================================

(define (http-get-function fnkey)
  (hash-table-ref/default *http-functions* fnkey (lambda () "nothing here yet")))

;; Main entry point to start a server. was start-server
(define (rmt:run hostn)

  ;;  ;; Configurations for server
  ;;  (tcp-buffer-size 2048)
  ;;  (max-connections 2048) 
  (debug:print 2 *default-log-port* "PID: "(current-process-id)". Attempting to start the server ...")
  (let* ((db              #f) ;;        (open-db)) ;; we don't want the server to be opening and closing the db unnecesarily
	 (hostname        (get-host-name))
	 (ipaddrstr       (let ((ipstr (if (string=? "-" hostn)
					   ;; (string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".")
					   (server:get-best-guess-address hostname)
					   #f)))
			    (if ipstr ipstr hostn))) ;; hostname))) 
	 (port            (portlogger:open-run-close portlogger:find-port))


	 ;; (link-tree-path  (common:get-linktree))
	 ;; (tmp-area        (common:get-db-tmp-area))
	 #;(start-file      (conc tmp-area "/.server-start")))
    (debug:print-info 0 *default-log-port* "portlogger recommended port: " port)
    (if *server-info*
	(begin


	  (servdat-host-set! *server-info* ipaddrstr)
	  (servdat-port-set! *server-info* port)
	  (servdat-status-set! *server-info* 'trying-port)
	  (servdat-trynum-set! *server-info* (+ (servdat-trynum *server-info*) 1)))
	(set! *server-info* (make-servdat host: ipaddrstr port: port)))

    (let* ((rep (rmt:try-start-server ipaddrstr port)))
      (let loop ((instr (nng-recv rep)))
	(let* ((data   (string->sexpr instr))
	       (res    (case data
			 ((quit) 'quit)
			 (else    (api:process-request *dbstruct-db* data))))
	       (resdat (sexpr->string res)))
	  (if (not (eq? res 'quit))
	      (begin
		(set! *db-last-access* (current-seconds))
		(nng-send rep resdat)

		(loop (nng-recv rep)))))))
    (debug:print-info 0 *default-log-port* "After server, should never see this")
    ;; server exit stuff here
    (let* ((portnum (servdat-port *server-info*))
	   (host    (servdat-host *server-info*)))
      (portlogger:open-run-close portlogger:set-port portnum "released")
      (if (not (equal? (get-host-name) host))
	  (debug:print-info 0 *default-log-port* "Server shutdown called for host "host", but we are on "(get-host-name))
	  (rmt:server-shutdown host portnum))
      ;; (bdat-time-to-exit-set! *bdat* #t) ;; tell on-exit to be fast as we've already cleaned up
      (portlogger:open-run-close portlogger:set-port port "released") ;; done in rmt:run
      ;; (debug:print-info 0 *default-log-port* "Max cached queries was    " *max-cache-size*)
      ;; (debug:print-info 0 *default-log-port* "Number of cached writes   " *number-of-writes*)
      ;; (debug:print-info 0 *default-log-port* "Average cached write time "
      ;; 		      (if (eq? *number-of-writes* 0)
      ;; 			  "n/a (no writes)"
      ;; 			  (/ *writes-total-delay*
      ;; 			     *number-of-writes*))
      ;; 		      " ms")
      ;; (debug:print-info 0 *default-log-port* "Number non-cached queries "  *number-non-write-queries*)
      ;; (debug:print-info 0 *default-log-port* "Average non-cached time   "
      ;; 		      (if (eq? *number-non-write-queries* 0)
      ;; 			  "n/a (no queries)"
      ;; 			  (/ *total-non-write-delay* 
      ;; 			     *number-non-write-queries*))
      ;; 		      " ms")
      
      (db:print-current-query-stats)
      (debug:print-info 0 *default-log-port* "Server shutdown complete. Exiting")
      )))

(define (rmt:try-start-server ipaddrstr portnum)
  (if *server-info* ;; update the server info as we might be trying next port
      (begin
	(servdat-host-set! *server-info* ipaddrstr)
	(servdat-port-set! *server-info* portnum)
	(servdat-status-set! *server-info* 'trying-port)
	(servdat-trynum-set! *server-info*
			     (+ (servdat-trynum *server-info*) 1)))
      (set! *server-info* (make-servdat host: ipaddrstr port: portnum)))
  (debug:print-info 0 *default-log-port* "rmt:try-start-server time="
		    (seconds->time-string (current-seconds))
		    " ipaddrsstr=" ipaddrstr
		    " portnum=" portnum)
  (assert (servdat? *server-info*) "FATAL: Must always have *server-info* properly set up by here.")
  (servdat-status-set! *server-info* 'starting)
  (servdat-port-set!   *server-info* portnum)
  (if (not (servdat-rep *server-info*))
      (let ((rep  (make-rep-socket)))
	(servdat-rep-set!    *server-info* rep)
	(socket-set! rep 'nng/recvtimeo 2000)))
  (let* ((rep (servdat-rep *server-info*)))
    (debug:print 0 *default-log-port* "INFO: Trying to start server on " ipaddrstr ":" portnum)
    (handle-exceptions
     exn
     (begin
       (print-error-message exn)
       (if (< portnum 64000)
	   (begin 
	     (debug:print 0 *default-log-port* "WARNING: attempt to start server failed. Trying again ...")
	     (debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
	     (debug:print 5 *default-log-port* "exn=" (condition->list exn))
	     (portlogger:open-run-close portlogger:set-failed portnum)
	     (debug:print 0 *default-log-port* "WARNING: failed to start on portnum: " portnum ", trying next port")
	     ;; (thread-sleep! 0.1)
	     (rmt:try-start-server ipaddrstr
				   (portlogger:open-run-close portlogger:find-port)))
	   (begin
	     (print "ERROR: Tried and tried but could not start the server, stopping at port "portnum))))
     (nng-listen rep (conc "tcp://*:" portnum))
     rep)))

;;======================================================================
;; S E R V E R   U T I L I T I E S 
;;======================================================================

;;======================================================================
;; C L I E N T S

;; only use for main.db - need to re-write some of this :(
;;
(define (get-lock-db sdat dbfile host port)
  (assert host "FATAL: get-lock-db called with host not set.")
  (assert port "FATAL: get-lock-db called with port not set.")
  (let* ((dbh (db:open-run-db dbfile db:initialize-db)) ;; open-run-db creates a standard db with schema used by all situations
	 (res (db:get-iam-server-lock dbh dbfile host port)))

    ;; res => list then already locked, check server is responsive
    ;;     => #t then sucessfully got the lock
    ;;     => #f reserved for future use as to indicate something went wrong
    (match res
      ((owner_pid owner_host owner_port event_time)
       (if (server-ready? owner_host owner_port "abc")
	   #f      ;; locked by someone else
	   (begin  ;; locked by someone dead and gone
	     (debug:print 0 *default-log-port* "WARNING: stale lock - have to steal it. This may fail.")
	     (db:steal-lock-db dbh dbfile port))))
      (#t  #t) ;; placeholder so that we don't touch res if it is #t
      (else (set! res #f)))
    (sqlite3:finalize! dbh)

	   (read-pkt->alist pkt-file pktspec: pktspec))
	 all-pkt-files)))

(define (server-address srv-pkt)
  (conc (alist-ref 'host srv-pkt) ":"
	(alist-ref 'port srv-pkt)))
	
(define (server-ready? host port key) ;; server-address is host:port
  (let* ((data (sexpr->string  `((cmd . ping)
				 (key . ,key)
				 (params . ()))))
	 (res  (open-send-receive-nn (conc host ":" port) data)))
    (if res
	(string->sexpr res)
	res)))

; from the pkts return servers associated with dbpath
;; NOTE: Only one can be alive - have to check on each
;;       in the list of pkts returned
;;
(define (get-viable-servers serv-pkts dbpath)
  (let loop ((tail serv-pkts)
	     (res  '()))
    (if (null? tail)
	res ;; NOTE: sort by age so oldest is considered first
	(let* ((spkt (car tail)))
	  (loop (cdr tail)
		(if (equal? dbpath (alist-ref 'dbpath spkt))
		    (cons spkt res)
		    res))))))

(define (remove-pkts-if-not-alive serv-pkts)
  (filter (lambda (pkt)
	    (let* ((host (alist-ref 'host pkt))
		   (port (alist-ref 'port pkt))
		   (key  (alist-ref 'servkey  pkt))
		   (pktz (alist-ref 'Z        pkt))
		   (res  (server-ready? host port key)))
	      (if res
		  res
		  (let* ((pktsdir (get-pkts-dir *toppath*))
			 (pktpath (conc pktsdir"/"pktz".pkt")))
		    (debug:print 0 *default-log-port* "WARNING: pkt with no server "pktpath)
		    (delete-file* pktpath)
		    #f))))
	  serv-pkts))

;; from viable servers get one that is alive and ready
;;
(define (get-the-server apath serv-pkts)
  (let loop ((tail serv-pkts))
    (if (null? tail)
	#f
	(let* ((spkt  (car tail))
	       (host  (alist-ref 'ipaddr spkt))
	       (port  (alist-ref 'port spkt))
	       (dbpth (alist-ref 'dbpath spkt))
	       (srvkey (alist-ref 'Z spkt)) ;; (alist-ref 'srvkey spkt))
	       (addr  (server-address spkt)))
	  (if (server-ready? host port srvkey)
	      spkt
	      (loop (cdr tail)))))))

;; am I the "first" in line server? I.e. my D card is smallest
;; use Z card as tie breaker
;;
(define (get-best-candidate serv-pkts dbpath)

	(debug:print-info 0 *default-log-port* "Waiting for server alive signature")
	(mutex-lock! *heartbeat-mutex*)
	(set! sdat *server-info*)
	(mutex-unlock! *heartbeat-mutex*)
	(if (and sdat
		 (not changed)
		 (> (- (current-seconds) start-time) 2))
	    (begin
	      (servdat-status-set! sdat 'iface-stable)
	      (debug:print-info 0 *default-log-port* "Received server alive signature, now attempting to lock in server")
	      ;; create a server pkt in *toppath*/.meta/srvpkts
	      
	      ;; TODO:
	      ;;   1. change sdat to stuct
	      ;;   2. add uuid to struct
	      ;;   3. update uuid in sdat here
	      ;;
	      (servdat-uuid-set! sdat
				 (register-server
				  pkts-dir *srvpktspec*
				  (get-host-name)
				  (servdat-port sdat) server-key
				  (servdat-host sdat) db-file))
	      ;; (set! *my-signature* (servdat-uuid sdat)) ;; replace with Z, no, stick with proper key
	      ;; now read pkts and see if we are a contender
	      (let* ((all-pkts     (get-all-server-pkts pkts-dir *srvpktspec*))
		     (viables      (get-viable-servers all-pkts db-file))
		     (alive        (remove-pkts-if-not-alive viables))
		     (best-srv     (get-best-candidate alive db-file))
		     (best-srv-key (if best-srv (alist-ref 'servkey best-srv) #f))
		     (i-am-srv     (equal? best-srv-key server-key))
		     (delete-pkt   (lambda ()
				     (let* ((pktfile (conc (get-pkts-dir *toppath*)
							 "/" (servdat-uuid *server-info*)
							 ".pkt")))

	 (server-start-time (current-seconds))
	 (pkts-dir          (get-pkts-dir))
	 (server-key        (rmt:get-signature)) ;; This servers key
	 (is-main           (equal? (args:get-arg "-db") ".db/main.db"))
	 (last-access       0)
	 (server-timeout    (server:expiration-timeout))
	 (shutdown-server-sequence (lambda (host port)
				     (set! *unclean-shutdown* #f)
				     (debug:print-info 0 *default-log-port* "Starting to shutdown the server. pid="(current-process-id))
				     (rmt:server-shutdown host port)
				     (portlogger:open-run-close portlogger:set-port port "released")
				     (exit)))
	 (timed-out?        (lambda ()
			      (<= (+ last-access server-timeout)
				  (current-seconds)))))
    (servdat-dbfile-set! *server-info* (args:get-arg "-db"))
    ;; main and run db servers have both got wait logic (could/should merge it)
    (if is-main
	(rmt:wait-for-server pkts-dir dbname server-key)
	(rmt:wait-for-stable-interface))
    ;; this is our forever loop
    (let* ((iface             (servdat-host *server-info*))
	   (port              (servdat-port *server-info*)))

      (let loop ((count          0)
		 (bad-sync-count 0)
		 (start-time     (current-milliseconds)))
	(if (and (not is-main)
		 (common:low-noise-print 60 "servdat-status"))
	    (debug:print-info 0 *default-log-port* "servdat-status is " (servdat-status *server-info*)))

	;; set up the database handle
	(mutex-lock! *heartbeat-mutex*)
	(if (not *dbstruct-db*) ;; no db opened yet, open the db and register with main if appropriate
	    (let ((watchdog (bdat-watchdog *bdat*)))		 
	      (debug:print 0 *default-log-port* "SERVER: dbprep")
	      (db:setup dbname) ;; sets *dbstruct-db* as side effect
	      (servdat-status-set! *server-info* 'db-opened)
	      ;; IFF I'm not main, call into main and register self
	      (if (not is-main)
		  (let ((res (rmt:register-server remdat
						  *toppath* iface port
						  server-key dbname)))
		    (if res ;; we are the server
			(servdat-status-set! *server-info* 'have-interface-and-db)
			;; now check that the db locker is alive, clear it out if not
			(let* ((serv-info (rmt:get-server-info *toppath* dbname)))
			  (match serv-info
			    ((host port servkey pid ipaddr apath dbpath)
			     (if (not (server-ready? host port servkey))
				 (begin
				   (debug:print-info 0 *default-log-port* "Server registered but not alive. Removing and trying again.")
				   (rmt:deregister-server remdat apath host port servkey dbpath) ;; servkey pid ipaddr apath dbpath)
				   
				   (loop (+ count 1) bad-sync-count start-time))))
			    (else
			     (debug:print 0 *default-log-port* "We are not the server for "dbname", exiting. Server info is: "serv-info)

	    (set! *unclean-shutdown* #f)
	    (debug:print-info 0 *default-log-port* "Server timed out. seconds since last db access: " (- (current-seconds) last-access))
	    (shutdown-server-sequence (get-host-name) port)
	    #;(debug:print-info 0 *default-log-port* "Sending 'quit to server, received: "
			      (open-send-receive-nn (conc iface":"port)      ;; do this here and not in server-shutdown
						    (sexpr->string 'quit)))
	    )))))))







;; Call this to start the actual server
;;
;; all routes though here end in exit ...
;;
;; This is the point at which servers are started
;;
(define (rmt:server-launch dbname)
  (debug:print-info 0 *default-log-port* "Entered rmt:server-launch")
  (let* ((th2 (make-thread (lambda ()
			     (debug:print-info 0 *default-log-port* "Server run thread started")
			     (rmt:run (if (args:get-arg "-server")
					  (args:get-arg "-server")
					  "-")
				      )) "Server run"))
	 (th3 (make-thread (lambda ()
			     (debug:print-info 0 *default-log-port* "Server monitor thread started")
			     (rmt:keep-running dbname)
			     "Keep running"))))
    (thread-start! th2)
    (thread-sleep! 0.252) ;; give the server time to settle before starting the keep-running monitor.
    (thread-start! th3)

	      (if (string-search (regexp (conc ":" port-num)) inl)
		  (begin
					;(print "Output: "  inl)
		    (set! ret  #t))
		  (loop (read-line inp)))))))
    ret))

;;start a server, returns the connection
;;
(define (start-nn-server portnum )
  (let ((rep (make-rep-socket))) ;; (nn-socket 'rep)))
    (socket-set! rep 'nng/recvtimeo 2000)
    (handle-exceptions ;; why have exception handler here? 
     exn
     (let ((emsg ((condition-property-accessor 'exn 'message) exn)))
       (print "ERROR: Failed to start server \"" emsg "\"")
       (exit 1))
      
     (nng-dial #;nn-bind rep (conc "tcp://*:" portnum)))
    rep))

(define (open-nn-connection host-port)
  (let ((req  (make-req-socket))
        (uri  (conc "tcp://" host-port)))
    (nng-dial req uri)
    (socket-set! req 'nng/recvtimeo 2000)
    req))

(define (send-receive-nn req msg)
  (nng-send req msg)
  (nng-recv req))

(define (close-nn-connection req)
  (nng-close! req))
  
;; ;; open connection to server, send message, close connection
;; ;;
;; (define (open-send-close-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
;;   (let ((req  (make-req-socket 'req))
;;         (uri  (conc "tcp://" host-port))

;;                                 (thread-terminate! th1))
;; 			      "timer thread")))
;;        (thread-start! th1)
;;        (thread-start! th2)
;;        (thread-join! th1)
;;        res))))
;; 
(define (open-send-receive-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
  (let ((req  (make-req-socket))
        (uri  (conc "tcp://" host-port))
        (res  #f)) 
    (handle-exceptions
     exn
     (let ((emsg ((condition-property-accessor 'exn 'message) exn)))
       ;; Send notification