Megatest

Diff
Login

Differences From Artifact [81fda153ef]:

To Artifact [be351e0e07]:


8
9
10
11
12
13
14
15

16
17
18
19
20
21
22
23
24
25

26
27
28




















29
30
31
32
33
34
35
36
37
38
39
40
41

42
43
44

45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

86
87
88
89
90
91
92
93
8
9
10
11
12
13
14

15
16
17
18
19
20
21
22
23
24

25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

61
62
63

64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

105
106
107
108
109
110
111
112
113







-
+









-
+



+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+












-
+


-
+




















-
+



















-
+








(define pub (make-socket 'pub))
(define pull (make-socket 'pull))

(bind-socket pub "tcp://*:5563")
(bind-socket pull "tcp://*:5564")

(define (open-db)
  (let* ((dbpath    "mockupserver.db")
  (let* ((dbpath    "mockup.db")
	 (dbexists  (file-exists? dbpath))
	 (db        (open-database dbpath)) ;; (never-give-up-open-db dbpath))
	 (handler   (make-busy-timeout 10)))
    (set-busy-handler! db handler)
    (if (not dbexists)
	(for-each
	 (lambda (stmt)
	   (execute db stmt))
	 (list
	  "CREATE TABLE clients (id INTEGER PRIMARY KEY,name TEXT,num_accesses INTEGER);"
	  "CREATE TABLE clients (id INTEGER PRIMARY KEY,name TEXT,num_accesses INTEGER DEFAULT 0);"
	  "CREATE TABLE vars    (var TEXT,val TEXT,CONSTRAINT vars_constraint UNIQUE (var));")))
    db))

(define cid-cache (make-hash-table))

(define (get-client-id db cname)
  (let ((cid (hash-table-ref/default cid-cache cname #f)))
    (if cid 
	cid
	(begin
	  (execute db "INSERT OR REPLACE INTO clients (name) VALUES(?);" cname)
	  (for-each-row 
	   (lambda (id)
	     (set! cid id))
	   db
	   "SELECT id FROM clients WHERE name=?;" cname)
	  (hash-table-set! cid-cache cname cid)
	  cid))))

(define (count-client db cname)
  (let ((cid (get-client-id db cname)))
    (execute db "UPDATE clients SET num_accesses=num_accesses+1 WHERE id=?;" cid)))

(define db (open-db))
;; (define queuelst '())
;; (define mx1 (make-mutex))

(define (process-queue queuelst)
  (for-each
   (lambda (item)
     (let ((cname (vector-ref item 1))
	   (clcmd (vector-ref item 2))
	   (cdata (vector-ref item 3)))
       (send-message pub cname send-more: #t)
       (send-message pub (case clcmd
			   ((setval)
			   ((set)
			    (apply execute db "INSERT OR REPLACE INTO vars (var,val) VALUES (?,?);" (string-split cdata))
			    "ok")
			   ((getval)
			   ((get)
			    (let ((res "noval"))
			      (for-each-row
			       (lambda (val)
				 (set! res val))
			       db 
			       "SELECT val FROM vars WHERE var=?;" cdata)
			      res))
			   (else (conc "unk cmd: " clcmd))))))
   queuelst))

(define th1 (make-thread 
	     (lambda ()
	       (let ((last-run 0)) ;; current-seconds when run last
		 (let loop ((queuelst '()))
		   (let* ((indat (receive-message* pull))
			  (parts (string-split indat ":"))
			  (cname (car parts))                   ;; client name
			  (clcmd (string->symbol (cadr parts))) ;; client cmd
			  (cdata (caddr parts))                 ;; client data
			  (svect (vector (current-seconds) cname clcmd cdata))) ;; record for the queue
		     ;; (print "Got indat=" indat)
		     (count-client db cname)
		     (case clcmd
		       ((sync) ;; just process the queue
			(print "Got sync from " cname)
			(process-queue queuelst)
			(loop '()))
		       ((imediate)
			(process-queue (cons svect queuelst))
			(loop '()))
		       (else
			(loop (cons svect queuelst))))))))
	     "server thread"))

(define push (make-socket 'push))
(connect-socket push "tcp://localhost:5564")

;; send a sync to the pull port
(define th2 (make-thread
	     (lambda ()
	       (let loop ()
		 (thread-sleep! 3)
		 (thread-sleep! 5)
		 ;; (print "Sending sync from server")
		 (send-message push "server:sync:nodat")
		 (loop)))
	     "sync thread"))

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