1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
|
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
|
-
-
-
+
+
+
|
(set! tmp (db:string->obj rawdat))
(vector-ref tmp 2))))
((zmq)
(handle-exceptions
exn
(begin
(thread-sleep! 5)
(if (> numretries 0)(apply cdb:client-call zmq-sockets qtype immediate (- numretries 1) params)))
(let* ((push-socket (vector-ref zmq-sockets 0))
(sub-socket (vector-ref zmq-sockets 1))
(if (> numretries 0)(apply cdb:client-call serverdat qtype immediate (- numretries 1) params)))
(let* ((push-socket (vector-ref serverdat 0))
(sub-socket (vector-ref serverdat 1))
(client-sig (server:get-client-signature))
(query-sig (message-digest-string (md5-primitive) (conc qtype immediate params)))
(zdat (db:obj->string (vector client-sig qtype immediate query-sig params (current-seconds)))) ;; (with-output-to-string (lambda ()(serialize params))))
(res #f)
(send-receive (lambda ()
(debug:print-info 11 "sending message")
(send-message push-socket zdat)
|
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
|
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
|
-
+
|
(if (> numretries 0)
(begin
(debug:print 2 "WARNING: no reply to query " params ", trying resend")
(debug:print-info 11 "re-sending message")
(send-message push-socket zdat)
(debug:print-info 11 "message re-sent")
(loop (- n 1)))
;; (apply cdb:client-call zmq-sockets qtype immediate (- numretries 1) params))
;; (apply cdb:client-call *runremote* qtype immediate (- numretries 1) params))
(begin
(debug:print 0 "ERROR: cdb:client-call timed out " params ", exiting.")
(exit 5))))))))
(debug:print-info 11 "Starting threads")
(let ((th1 (make-thread send-receive "send receive"))
(th2 (make-thread timeout "timeout")))
(thread-start! th1)
|