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
|
directory-utils
ulex
)
(define help "Usage: ulex-test COMMAND
where COMMAND is one of:
do-test : run the basic req/rep test
run tcp://host:port : start test server - start several in same dir
(define address-tcp-1 "tcp://localhost:5555")
(define address-tcp-2 "tcp://localhost:6666")
(define address-inproc-1 "inproc://local1")
(define address-inproc-2 "inproc://local2")
;;;
;;; Req-Rep
;;;
(define (make-listening-reply-socket address)
(let ((socket (make-rep-socket)))
(socket-set! socket 'nng/recvtimeo 2000)
(nng-listen socket address)
socket))
(define (make-dialed-request-socket address)
(let ((socket (make-req-socket)))
(socket-set! socket 'nng/recvtimeo 2000)
(nng-dial socket address)
socket))
(define (req-rep-test address)
(let ((rep (make-listening-reply-socket address))
(req (make-dialed-request-socket address)))
(nng-send req "message 1")
(nng-recv rep)
(nng-send rep "message")
(begin0
(nng-recv req)
(nng-close! rep))))
o-test)
(test-group "nng"
(test "tcp req-rep"
"message"
(req-rep-test address-tcp-1))
(test "inproc req-rep"
"message"
(req-rep-test address-inproc-1)))
(test-exit))
;; this should be run in a thread
(define (run-listener-responder socket myaddr)
(let loop ((status 'running))
(let* ((msg (nng-recv socket))
(response (process-message msg)))
(if (not (eq? response 'done))
(begin
(nng-send socket response)
(loop status))))))
(define *channels* (make-hash-table))
(define (call channels msg addr)
(let* ((csocket (hash-table-ref/default channels addr #f))
(socket (or csocket (make-dialed-request-socket addr))))
(nng-send socket msg)
(print "Sent: "msg", received: "(nng-recv socket))
(if (not (hash-table-exists? channels addr))
(hash-table-set! channels addr socket))))
|
<
|
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
|
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
<
|
<
|
|
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
|
directory-utils
ulex
)
(define help "Usage: ulex-test COMMAND
where COMMAND is one of:
run host:port : start test server - start several in same dir
")
(define (call uconn msg addr)
(print "Sent: "msg", received: "
(send-receive uconn addr 'hello msg)))
;; start => hello 0
;; hello 0 => hello 1
;; hello 1 => hello 2
;; ...
;; hello 11 => 'done
;;
|
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
|
(conc msg " 0"))
(else
"hello 0"))))
(define (main)
(match
(command-line-arguments)
(("do-test")(do-test))
((run myaddr)
(let* ((endtimes (+ (current-seconds) 20)) ;; run for 20 seconds
(socket (make-listening-reply-socket myaddr))
(rfile (conc ".runners/"(get-host-name)"-"(current-process-id)))
(th1 (make-thread (lambda ()
(run-listener-responder socket myaddr)
)
"responder")))
(if (not (and (file-exists? ".runners")
(directory? ".runners")))
(create-directory ".runners" #t))
(with-output-to-file rfile
(lambda ()
(print myaddr)))
(thread-start! th1)
(let loop ((entries '()))
(if (> (current-seconds) endtimes)
(begin
(delete-file* rfile)
(sleep 1)
(exit))
(if (null? entries)
(loop (glob ".runners/*"))
(let* ((entry (car entries))
(destaddr (with-input-from-file entry read-line)))
(call *channels* (conc "hello-from-"destaddr) destaddr)
;; (thread-sleep! 0.025)
(loop (cdr entries))))))))
((cmd)(print "ERROR: command "cmd", not recognised.\n\n"help))
(else
(print help))))
) ;; end module
(import nng-test)
(main)
|
<
|
>
|
>
>
|
|
<
<
<
<
|
<
|
|
|
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
|
(conc msg " 0"))
(else
"hello 0"))))
(define (main)
(match
(command-line-arguments)
((run myport)
;; start listener
;; put myaddr into file by host-pid in .runners
;; for 1 minute
;; get all in .runners
;; call each with a message
;;
(let* ((port (string->number myport))
(endtimes (+ (current-seconds) 20)) ;; run for 20 seconds
(handler (lambda (rem-host-port qrykey cmd params)
(process-message params)))
(uconn (run-listener handler myport))
(rfile (conc ".runners/"(get-host-name)"-"(current-process-id))))
(if (not (and (file-exists? ".runners")
(directory? ".runners")))
(create-directory ".runners" #t))
(with-output-to-file rfile
(lambda ()
(print myport)))
(let loop ((entries '()))
(if (> (current-seconds) endtimes)
(begin
(delete-file* rfile)
(sleep 1)
(exit))
(if (null? entries)
(loop (glob ".runners/*"))
(let* ((entry (car entries))
(destaddr (with-input-from-file entry read-line)))
(call uconn (conc "hello-from-"myport"to-"destaddr) destaddr)
;; (thread-sleep! 0.025)
(loop (cdr entries))))))))
((cmd)(print "ERROR: command "cmd", not recognised.\n\n"help))
(else
(print help))))
) ;; end module
(import ulex-test)
(main)
|