1 # vim: set et nosta sw=4 ts=4 ft=nim : |
|
2 # |
|
3 # Copyright (c) 2016, Mahlon E. Smith <mahlon@martini.nu> |
|
4 # All rights reserved. |
|
5 # Redistribution and use in source and binary forms, with or without |
|
6 # modification, are permitted provided that the following conditions are met: |
|
7 # |
|
8 # * Redistributions of source code must retain the above copyright |
|
9 # notice, this list of conditions and the following disclaimer. |
|
10 # |
|
11 # * Redistributions in binary form must reproduce the above copyright |
|
12 # notice, this list of conditions and the following disclaimer in the |
|
13 # documentation and/or other materials provided with the distribution. |
|
14 # |
|
15 # * Neither the name of Mahlon E. Smith nor the names of his |
|
16 # contributors may be used to endorse or promote products derived |
|
17 # from this software without specific prior written permission. |
|
18 # |
|
19 # THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND ANY |
|
20 # EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED |
|
21 # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE |
|
22 # DISCLAIMED. IN NO EVENT SHALL THE REGENTS AND CONTRIBUTORS BE LIABLE FOR ANY |
|
23 # DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES |
|
24 # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; |
|
25 # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND |
|
26 # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
|
27 # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS |
|
28 # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
|
29 |
|
30 ## Overview |
|
31 ## ============ |
|
32 ## |
|
33 ## This is a pure-nim client library for interacting with Stomp 1.2 |
|
34 ## compliant messaging brokers. |
|
35 ## |
|
36 ## https://stomp.github.io/stomp-specification-1.2.html |
|
37 ## |
|
38 ## Stomp is a simple protocol for message passing between clients, using a central |
|
39 ## broker. It is a subset of other more elaborate protocols (like AMQP), supporting |
|
40 ## only the most used features of common brokers. |
|
41 ## |
|
42 ## Because this library is pure-nim, there are no external dependencies. If you |
|
43 ## can compile a nim binary, you can participate in advanced messaging between processes. |
|
44 ## |
|
45 ## A list of broker support for Stomp can be found here: |
|
46 ## https://stomp.github.io/implementations.html. |
|
47 ## |
|
48 ## This library has been tested with recent versions of RabbitMQ. If it |
|
49 ## works for you with another broker, please let the author know. |
|
50 ## |
|
51 ## |
|
52 ## Protocol support |
|
53 ## ---------------- |
|
54 ## |
|
55 ## Examples |
|
56 ## ========= |
|
57 ## |
|
58 ## Connecting with SSL |
|
59 ## ------------------- |
|
60 ## |
|
61 |
|
62 import |
|
63 strutils, |
|
64 nativesockets, |
|
65 net, |
|
66 os, |
|
67 times, |
|
68 uri |
|
69 |
|
70 const |
|
71 VERSION = "0.1.0" ## The current program version. |
|
72 NULL = "\x00" ## The NULL character. |
|
73 CR = "\r" ## The carriage return character. |
|
74 CRLF = "\r\n" ## Carriage return + Line feed (EOL). |
|
75 |
|
76 |
|
77 # Exceptions |
|
78 # |
|
79 type |
|
80 StompError* = object of ValueError ## A generic Stomp error state. |
|
81 |
|
82 StompClient* = ref object of RootObj ## An object that represents a connection to a Stomp compatible server. |
|
83 socket: Socket ## The socket object attached to this client. |
|
84 connected*: bool ## Is the client currently connected? |
|
85 uri*: Uri ## The URI used to instantiate this client. |
|
86 username: string ## The Stomp server user, if any. |
|
87 password: string ## The Stomp server password, if any. |
|
88 host: string ## The host or IP address to connect to. |
|
89 port: Port ## Optional, if Stomp is on a non-default port. |
|
90 vhost: string ## Parsed from the URI path, a Stomp "virtual host". |
|
91 timeout*: int ## Global socket timeout. |
|
92 last_msgtime*: Time ## Timestamp of last seen server message. |
|
93 options: tuple[ heartbeat: int ] ## Any supported client options, derived from the URI query string. |
|
94 subscriptions*: seq[ string ] ## Registered client subscriptions. Array position is the ID. |
|
95 transactions*: seq[ string ] ## Any currently open transactions. |
|
96 serverinfo: seq[ tuple[name: string, value: string] ] ## Server metadata, populated upon a successful connection. |
|
97 |
|
98 connected_callback*: proc ( client: StompClient, response: StompResponse ): void |
|
99 error_callback*: proc ( client: StompClient, response: StompResponse ): void |
|
100 heartbeat_callback*: proc ( client: StompClient, response: StompResponse ): void |
|
101 message_callback*: proc ( client: StompClient, response: StompResponse ): void |
|
102 missed_heartbeat_callback*: proc ( client: StompClient ): void |
|
103 receipt_callback*: proc ( client: StompClient, response: StompResponse ): void |
|
104 |
|
105 StompResponse* = ref object of RootObj ## A parsed packet from a Stomp server. |
|
106 headers: seq[ tuple[name: string, value: string] ] ## Any headers in the response. Access with the `[]` proc. |
|
107 frame*: string ## The Stomp frame type. |
|
108 payload*: string ## The message body, if any. |
|
109 |
|
110 |
|
111 # convenience |
|
112 proc printf( formatstr: cstring ) {.header: "<stdio.h>", varargs.} |
|
113 |
|
114 |
|
115 #------------------------------------------------------------------- |
|
116 # R E S P O N S E |
|
117 #------------------------------------------------------------------- |
|
118 |
|
119 proc is_eol( s: string ): bool = |
|
120 ## Convenience method, returns **true** if string is a Stomp EOF. |
|
121 return s == CR or s == CRLF |
|
122 |
|
123 |
|
124 proc parse_headers( response: StompResponse, c: StompClient ): int = |
|
125 ## Parse response headers from a stream. |
|
126 ## Returns the content length of the response body, or 0. |
|
127 result = 0 |
|
128 var line = "" |
|
129 |
|
130 c.socket.readline( line, c.timeout ) |
|
131 while not line.is_eol: |
|
132 if defined( debug ): printf " <-- %s\n", line |
|
133 var header = line.split( ":" ) |
|
134 if header.len < 2: break |
|
135 response.headers.add( (header[0], header[1]) ) |
|
136 if cmpIgnoreCase( header[0], "content-length" ) == 0: result = header[1].parse_int |
|
137 c.socket.readline( line, c.timeout ) |
|
138 |
|
139 |
|
140 proc parse_payload( response: StompResponse, c: StompClient, bodylength = 0 ): void = |
|
141 ## Parse message payload from a stream. |
|
142 let bufsize = 8192 |
|
143 var |
|
144 buf = "" |
|
145 data = "" |
|
146 |
|
147 |
|
148 # If we already know the length of the body, just perform a buffered read. |
|
149 # |
|
150 if bodylength > 0: |
|
151 var |
|
152 readtotal = 0 |
|
153 readamt = 0 |
|
154 remaining = 0 |
|
155 |
|
156 while readtotal != bodylength: |
|
157 remaining = bodylength - readtotal |
|
158 |
|
159 if remaining < bufsize: |
|
160 readamt = remaining |
|
161 else: |
|
162 readamt = bufsize |
|
163 |
|
164 buf = newString( readamt ) |
|
165 readtotal = readtotal + c.socket.recv( buf, readamt, c.timeout ) |
|
166 data = data & buf |
|
167 |
|
168 # Eat the NULL terminator. |
|
169 discard c.socket.recv( buf, 1, c.timeout ) |
|
170 |
|
171 # Inefficient path. |
|
172 # |
|
173 else: |
|
174 while buf != NULL: |
|
175 discard c.socket.recv( buf, 1, c.timeout ) |
|
176 data = data & buf |
|
177 |
|
178 response.payload = data |
|
179 |
|
180 |
|
181 proc newStompResponse( c: StompClient ): StompResponse = |
|
182 ## Initialize a response object, which parses and contains |
|
183 ## the Stomp headers and any additional important data from |
|
184 ## the broker socket. |
|
185 new( result ) |
|
186 result.headers = @[] |
|
187 |
|
188 # Get the frame type, record last seen server activity time. |
|
189 # |
|
190 var line = "" |
|
191 c.socket.readline( line, c.timeout ) |
|
192 c.last_msgtime = get_time() |
|
193 |
|
194 # Heartbeat packets (empties.) |
|
195 # |
|
196 # This could -also- parse optional EOLs from the prior |
|
197 # message (after the NULL separator), but since it is a no-op, |
|
198 # this seems harmless. |
|
199 # |
|
200 if line.is_eol: |
|
201 result.frame = "HEARTBEAT" |
|
202 return result |
|
203 |
|
204 # All other types. |
|
205 # |
|
206 result.frame = line |
|
207 if defined( debug ): |
|
208 printf " <-- %s\n", line |
|
209 |
|
210 # Parse headers and body. |
|
211 # |
|
212 var length = result.parse_headers( c ) |
|
213 if result.frame == "MESSAGE" or result.frame == "ERROR": |
|
214 result.parse_payload( c, length ) |
|
215 |
|
216 # If the response -could- have a body, the NULL has already |
|
217 # been removed from the stream while we checked for one. |
|
218 # |
|
219 if result.payload.len > 0: |
|
220 if result.payload == NULL: # We checked for a body, but there was none. |
|
221 result.payload = nil |
|
222 if defined( debug ): printf " <--\n <-- ^@\n\n" |
|
223 else: |
|
224 if defined( debug ): printf " <--\n <-- (payload)^@\n\n" |
|
225 |
|
226 # Otherwise, pop off the NULL terminator now. |
|
227 # |
|
228 else: |
|
229 discard c.socket.recv( line, 1, c.timeout ) |
|
230 if defined( debug ): printf " <--\n <-- ^@\n\n" |
|
231 |
|
232 |
|
233 proc `$`*( r: StompResponse ): string = |
|
234 ## Represent a Stomp response as a string. |
|
235 result = r.frame & ": " & $r.headers |
|
236 |
|
237 |
|
238 proc `[]`*( response: StompResponse, key: string ): string = |
|
239 ## Get a specific header from a Stomp response. |
|
240 for header in response.headers: |
|
241 if cmpIgnoreCase( key, header.name ) == 0: |
|
242 return header.value |
|
243 return nil |
|
244 |
|
245 |
|
246 #------------------------------------------------------------------- |
|
247 # C A L L B A C K S |
|
248 #------------------------------------------------------------------- |
|
249 |
|
250 proc default_error_callback( c: StompClient, response: StompResponse ) = |
|
251 ## Something bad happened. Disconnect from the server, build an error message, |
|
252 ## and raise an exception. |
|
253 c.socket.close |
|
254 c.connected = false |
|
255 |
|
256 var detail = response.payload |
|
257 var msg = response[ "message" ] |
|
258 if $detail[ ^1 ] == "\n": detail = detail[ 0 .. ^2 ] # chomp |
|
259 |
|
260 if detail.len > 0: msg = msg & " (" & detail & ")" |
|
261 raise newException( StompError, "ERROR: " & msg ) |
|
262 |
|
263 |
|
264 proc default_missed_heartbeat_callback( c: StompClient ) = |
|
265 ## Timeout while connected to the broker. |
|
266 c.socket.close |
|
267 c.connected = false |
|
268 raise newException( StompError, "Heartbeat timeout. Last activity: " & $c.last_msgtime ) |
|
269 |
|
270 |
|
271 |
|
272 #------------------------------------------------------------------- |
|
273 # C L I E N T |
|
274 #------------------------------------------------------------------- |
|
275 |
|
276 proc newStompClient*( s: Socket, uri: string ): StompClient = |
|
277 ## Create a new Stomp client object from a preexisting **socket**, |
|
278 ## and a stomp **URI** string. |
|
279 ## |
|
280 ## .. code-block:: nim |
|
281 ## |
|
282 ## var socket = newSocket() |
|
283 ## var stomp = newStompClient( socket, "stomp://test:test@example.com/%2Fvhost" ) |
|
284 ## |
|
285 ## or if connecting with SSL, when compiled with -d:ssl: |
|
286 ## |
|
287 ## .. code-block:: nim |
|
288 ## |
|
289 ## var socket = newSocket() |
|
290 ## let sslContext = newContext( verifyMode = CVerifyNone ) |
|
291 ## sslContext.wrapSocket(socket) |
|
292 ## var stomp = newStompClient( socket, "stomp+ssl://test:test@example.com/%2Fvhost" ) |
|
293 ## |
|
294 new( result ) |
|
295 result.socket = s |
|
296 result.connected = false |
|
297 result.uri = parse_uri( uri ) |
|
298 result.username = result.uri.username |
|
299 result.password = result.uri.password |
|
300 result.host = result.uri.hostname |
|
301 result.vhost = result.uri.path |
|
302 result.timeout = 500 |
|
303 result.subscriptions = @[] |
|
304 result.transactions = @[] |
|
305 |
|
306 # Parse any supported options in the query string. |
|
307 # |
|
308 for pairs in result.uri.query.split( '&' ): |
|
309 let opt = pairs.split( '=' ) |
|
310 try: |
|
311 case opt[0]: |
|
312 of "heartbeat": |
|
313 result.options.heartbeat = opt[1].parse_int |
|
314 else: |
|
315 discard |
|
316 except IndexError, ValueError: |
|
317 discard |
|
318 |
|
319 # Set default STOMP port if otherwise unset. |
|
320 # |
|
321 if not result.uri.scheme.contains( "stomp" ): |
|
322 raise newException( StompError, "Unknown scheme: " & result.uri.scheme ) |
|
323 var port: int |
|
324 if result.uri.port == "": |
|
325 if result.uri.scheme.contains( "+ssl" ): |
|
326 port = 61614 |
|
327 else: |
|
328 port = 61613 |
|
329 else: |
|
330 port = result.uri.port.parse_int |
|
331 |
|
332 result.port = Port( port ) |
|
333 |
|
334 # Decode URI encoded slashes for vhosts. |
|
335 result.vhost = result.vhost.replace( "%2f", "/" ).replace( "%2F", "/" ).replace( "//", "/" ) |
|
336 |
|
337 |
|
338 proc socksend( c: StompClient, data: string ): void = |
|
339 ## Send data on the connected socket with optional debug output. |
|
340 c.socket.send( data ) |
|
341 if defined( debug ): printf " --> %s", data |
|
342 |
|
343 |
|
344 proc finmsg( c: StompClient ): void = |
|
345 ## Send data on the connected socket with optional debug output. |
|
346 c.socket.send( CRLF & NULL & CRLF ) |
|
347 if defined( debug ): printf " --> \n --> ^@\n\n" |
|
348 |
|
349 |
|
350 proc `[]`*( c: StompClient, key: string ): string = |
|
351 ## Get a specific value from the server metadata, set during the initial connection. |
|
352 if not c.connected: return nil |
|
353 for header in c.serverinfo: |
|
354 if cmpIgnoreCase( key, header.name ) == 0: |
|
355 return header.value |
|
356 return nil |
|
357 |
|
358 |
|
359 proc `$`*( c: StompClient ): string = |
|
360 ## Represent the stomp client as a string, after masking the password. |
|
361 let uri = ( $c.uri ).replace( ":" & c.uri.password & "@", "@" ) |
|
362 result = "(NimStomp v" & VERSION & ( if c.connected: " connected" else: " not connected" ) & " to " & uri |
|
363 if not c[ "server" ].is_nil: |
|
364 result.add( " --> " & c["server"] ) |
|
365 result.add( ")" ) |
|
366 |
|
367 |
|
368 proc connect*( c: StompClient ): void = |
|
369 ## Establish a connection to the Stomp server. |
|
370 if c.connected: return |
|
371 |
|
372 var headers: seq[ tuple[name: string, value: string] ] = @[] |
|
373 headers.add( ("accept-version", "1.2") ) |
|
374 |
|
375 # Stomp 1.2 requires the Host: header. Use the path as a vhost if |
|
376 # supplied, otherwise use the hostname of the server. |
|
377 # |
|
378 if c.vhost != "": |
|
379 headers.add( ("host", c.vhost) ) |
|
380 else: |
|
381 headers.add( ("host", c.host) ) |
|
382 |
|
383 if c.username != "": headers.add( ("login", c.username) ) |
|
384 if c.password != "": headers.add( ("passcode", c.password) ) |
|
385 if c.options.heartbeat > 0: |
|
386 let heartbeat = c.options.heartbeat * 1000 |
|
387 headers.add( ("heart-beat", "0," & $heartbeat) ) |
|
388 |
|
389 # Connect the socket and send the headers off. |
|
390 # |
|
391 c.socket.connect( c.host, c.port ) |
|
392 c.socksend( "CONNECT" & CRLF ) |
|
393 for header in headers: |
|
394 c.socksend( header.name & ":" & header.value & CRLF ) |
|
395 c.finmsg |
|
396 |
|
397 # Retreive and copy server metadata to client object. |
|
398 # |
|
399 var response = newStompResponse( c ) |
|
400 c.serverinfo = response.headers |
|
401 |
|
402 if response.frame != "CONNECTED": |
|
403 if not isNil( c.error_callback ): |
|
404 c.error_callback( c, response ) |
|
405 else: |
|
406 c.default_error_callback( response ) |
|
407 else: |
|
408 c.connected = true |
|
409 if not isNil( c.connected_callback ): |
|
410 c.connected_callback( c, response ) |
|
411 |
|
412 |
|
413 proc disconnect*( c: StompClient ): void = |
|
414 ## Break down the connection to the Stomp server nicely. |
|
415 if not c.connected: return |
|
416 c.socksend( "DISCONNECT" & CRLF ) |
|
417 c.finmsg |
|
418 |
|
419 c.socket.close |
|
420 c.connected = false |
|
421 |
|
422 |
|
423 proc add_txn( c: StompClient ): void = |
|
424 ## Add a transaction header if there is only a single open txn. |
|
425 if c.transactions.len != 1: return |
|
426 c.socksend( "transaction:" & c.transactions[0] & CRLF ) |
|
427 |
|
428 |
|
429 proc send*( c: StompClient, |
|
430 destination: string, |
|
431 message: string = nil, |
|
432 contenttype: string = nil, |
|
433 headers: seq[ tuple[name: string, value: string] ] = @[] ): void = |
|
434 ## Send a **message** to **destination**. |
|
435 ## |
|
436 ## A Content-Length header is automatically and always included. |
|
437 ## A **contenttype** is optional, but strongly recommended. |
|
438 ## |
|
439 ## Additionally, a transaction ID is automatically added if there is only |
|
440 ## one transaction active. If you need to attach this message to a particular |
|
441 ## transaction ID, you'll need to add it yourself with the user defined |
|
442 ## **headers**. |
|
443 |
|
444 if not c.connected: raise newException( StompError, "Client is not connected." ) |
|
445 c.socksend( "SEND" & CRLF ) |
|
446 c.socksend( "destination:" & destination & CRLF ) |
|
447 c.socksend( "content-length:" & $message.len & CRLF ) |
|
448 if not contenttype.is_nil: c.socksend( "content-type:" & contenttype & CRLF ) |
|
449 |
|
450 # Add custom headers. Add transaction header if one isn't manually |
|
451 # present (and a transaction is open.) |
|
452 # |
|
453 var txn_seen = false |
|
454 for header in headers: |
|
455 if header.name == "transaction": txn_seen = true |
|
456 c.socksend( header.name & ":" & header.value & CRLF ) |
|
457 if not txn_seen: c.add_txn |
|
458 |
|
459 if message.is_nil: |
|
460 c.finmsg |
|
461 else: |
|
462 c.socket.send( CRLF & message & NULL ) |
|
463 if defined( debug ): printf " -->\n --> (payload)^@\n\n" |
|
464 |
|
465 |
|
466 proc subscribe*( c: StompClient, |
|
467 destination: string, |
|
468 ack = "auto", |
|
469 headers: seq[ tuple[name: string, value: string] ] = @[] ): void = |
|
470 ## Subscribe to messages at **destination**. |
|
471 ## |
|
472 ## Setting **ack** to "client" or "client-individual" enables client ACK/NACK mode. |
|
473 ## In this mode, incoming messages aren't considered processed by |
|
474 ## the server unless they receive ACK. By default, the server |
|
475 ## considers the message processed if a client simply accepts it. |
|
476 ## |
|
477 ## You may optionally add any additional **headers** the server may support. |
|
478 |
|
479 if not c.connected: raise newException( StompError, "Client is not connected." ) |
|
480 c.socksend( "SUBSCRIBE" & CRLF ) |
|
481 c.socksend( "destination:" & destination & CRLF ) |
|
482 c.socksend( "id:" & $c.subscriptions.len & CRLF ) |
|
483 if ack == "client" or ack == "client-individual": |
|
484 c.socksend( "ack:" & ack & CRLF ) |
|
485 else: |
|
486 if ack != "auto": raise newException( StompError, "Unknown ack type: " & ack ) |
|
487 |
|
488 for header in headers: |
|
489 c.socksend( header.name & ":" & header.value & CRLF ) |
|
490 c.finmsg |
|
491 c.subscriptions.add( destination ) |
|
492 |
|
493 |
|
494 proc unsubscribe*( c: StompClient, |
|
495 destination: string, |
|
496 headers: seq[ tuple[name: string, value: string] ] = @[] ): void = |
|
497 ## Unsubscribe from messages at **destination**. |
|
498 ## You may optionally add any additional **headers** the server may support. |
|
499 |
|
500 if not c.connected: raise newException( StompError, "Client is not connected." ) |
|
501 var |
|
502 sub_id: int |
|
503 i = 0 |
|
504 |
|
505 # Find the ID of the subscription. |
|
506 # |
|
507 for sub in c.subscriptions: |
|
508 if sub == destination: |
|
509 sub_id = i |
|
510 break |
|
511 i = i + 1 |
|
512 |
|
513 c.socksend( "UNSUBSCRIBE" & CRLF ) |
|
514 c.socksend( "id:" & $sub_id & CRLF ) |
|
515 for header in headers: |
|
516 c.socksend( header.name & ":" & header.value & CRLF ) |
|
517 c.finmsg |
|
518 c.subscriptions[ sub_id ] = "" |
|
519 |
|
520 |
|
521 proc begin*( c: StompClient, txn: string ): void = |
|
522 ## Begin a new transaction on the broker, using **txn** as the identifier. |
|
523 c.socksend( "BEGIN" & CRLF ) |
|
524 c.socksend( "transaction:" & txn & CRLF ) |
|
525 c.finmsg |
|
526 c.transactions.add( txn ) |
|
527 |
|
528 |
|
529 proc commit*( c: StompClient, txn: string = nil ): void = |
|
530 ## Finish a specific transaction **txn**, or the most current if unspecified. |
|
531 var transaction = txn |
|
532 if transaction.is_nil and c.transactions.len > 0: transaction = c.transactions.pop |
|
533 if transaction.is_nil: return |
|
534 |
|
535 c.socksend( "COMMIT" & CRLF ) |
|
536 c.socksend( "transaction:" & transaction & CRLF ) |
|
537 c.finmsg |
|
538 |
|
539 # Remove the transaction from the queue. |
|
540 # |
|
541 var new_transactions: seq[ string ] = @[] |
|
542 for txn in c.transactions: |
|
543 if txn != transaction: new_transactions.add( txn ) |
|
544 c.transactions = new_transactions |
|
545 |
|
546 |
|
547 proc abort*( c: StompClient, txn: string = nil ): void = |
|
548 ## Cancel a specific transaction **txn**, or the most current if unspecified. |
|
549 var transaction = txn |
|
550 if transaction.is_nil and c.transactions.len > 0: transaction = c.transactions.pop |
|
551 if transaction.is_nil: return |
|
552 |
|
553 c.socksend( "ABORT" & CRLF ) |
|
554 c.socksend( "transaction:" & transaction & CRLF ) |
|
555 c.finmsg |
|
556 |
|
557 # Remove the transaction from the queue. |
|
558 # |
|
559 var new_transactions: seq[ string ] = @[] |
|
560 for txn in c.transactions: |
|
561 if txn != transaction: new_transactions.add( txn ) |
|
562 c.transactions = new_transactions |
|
563 |
|
564 |
|
565 proc ack*( c: StompClient, id: string, transaction: string = nil ): void = |
|
566 ## Acknowledge message **id**. Optionally, attach this acknowledgement |
|
567 ## to a specific **transaction** -- if there's only one active, it is |
|
568 ## added automatically. |
|
569 c.socksend( "ACK" & CRLF ) |
|
570 c.socksend( "id:" & id & CRLF ) |
|
571 if not transaction.is_nil: |
|
572 c.socksend( "transaction:" & transaction & CRLF ) |
|
573 else: |
|
574 c.add_txn |
|
575 c.finmsg |
|
576 |
|
577 |
|
578 proc nack*( c: StompClient, id: string, transaction: string = nil ): void = |
|
579 ## Reject message **id**. Optionally, attach this rejection to a |
|
580 ## specific **transaction** -- if there's only one active, it is |
|
581 ## added automatically. |
|
582 ## |
|
583 ## Subscribe to a queue with ACK mode enabled, and reject the message |
|
584 ## on error: |
|
585 ## |
|
586 ## .. code-block:: nim |
|
587 ## |
|
588 ## stomp.subscribe( "/queue/test", "client-individual" ) |
|
589 ## FIXME: attach procs |
|
590 ## stomp.wait_for_messages |
|
591 ## |
|
592 c.socksend( "NACK" & CRLF ) |
|
593 c.socksend( "id:" & id & CRLF ) |
|
594 if not transaction.is_nil: |
|
595 c.socksend( "transaction:" & transaction & CRLF ) |
|
596 else: |
|
597 c.add_txn |
|
598 c.finmsg |
|
599 |
|
600 |
|
601 proc wait_for_messages*( c: StompClient, loop=true ) = |
|
602 ## Enter a blocking select loop, dispatching to the appropriate proc |
|
603 ## for the received message type. Return after a single message |
|
604 ## is received if **loop** is set to **false**. |
|
605 |
|
606 if not c.connected: raise newException( StompError, "Client is not connected." ) |
|
607 |
|
608 while true: |
|
609 var |
|
610 timeout: int |
|
611 fds = @[ c.socket.get_fd ] |
|
612 |
|
613 # Check for missed heartbeats, with an additional second |
|
614 # of wiggle-room. |
|
615 # |
|
616 if c.options.heartbeat > 0: |
|
617 timeout = ( c.options.heartbeat + 1 ) * 1000 |
|
618 else: |
|
619 timeout = -1 |
|
620 |
|
621 if select( fds, timeout ) == 0: # timeout, only happens if heartbeating missed |
|
622 if not isNil( c.missed_heartbeat_callback ): |
|
623 c.missed_heartbeat_callback( c ) |
|
624 else: |
|
625 c.default_missed_heartbeat_callback |
|
626 if loop: continue else: break |
|
627 |
|
628 let response = newStompResponse( c ) |
|
629 case response.frame: |
|
630 |
|
631 of "HEARTBEAT": |
|
632 if not isNil( c.heartbeat_callback ): |
|
633 c.heartbeat_callback( c, response ) |
|
634 continue |
|
635 |
|
636 of "RECEIPT": |
|
637 if not isNil( c.receipt_callback ): |
|
638 c.receipt_callback( c, response ) |
|
639 |
|
640 of "MESSAGE": |
|
641 if not isNil( c.message_callback ): |
|
642 c.message_callback( c, response ) |
|
643 |
|
644 of "ERROR": |
|
645 if not isNil( c.error_callback ): |
|
646 c.error_callback( c, response ) |
|
647 else: |
|
648 c.default_error_callback( response ) |
|
649 |
|
650 else: |
|
651 if defined( debug ): |
|
652 echo "Strange broker frame: " & response.repr |
|
653 |
|
654 if not loop: break |
|
655 |
|
656 |
|
657 |
|
658 #------------------------------------------------------------------- |
|
659 # T E S T S |
|
660 #------------------------------------------------------------------- |
|
661 |
|
662 # Functional (rather than unit) tests. Requires a Stomp compatible broker. |
|
663 # This was tested against RabbitMQ 3.5.3 and 3.6.0. |
|
664 # 3.6.0 was -so- much faster. |
|
665 # |
|
666 # First start up a message receiver: |
|
667 # ./stomp receiver [stomp-uri] [subscription-destination] |
|
668 # |
|
669 # then run another process, to publish stuff: |
|
670 # ./stomp publisher [stomp-uri] [publish-destination] |
|
671 # |
|
672 # An example with an AMQP "direct" exchange, and an exclusive queue: |
|
673 # ./stomp publisher stomp://test:test@localhost/?heartbeat=10 /exchange/test |
|
674 # ./stomp receiver stomp://test:test@localhost/?heartbeat=10 /exchange/test |
|
675 # |
|
676 # Then just let 'er run. |
|
677 # |
|
678 # You can also run a nieve benchmark (deliveries/sec): |
|
679 # |
|
680 # ./stomp benchmark stomp://test:test@localhost/ /exchange/test |
|
681 # |
|
682 # It will set messages to require acknowledgement, and nack everything, causing |
|
683 # a delivery loop for 10 seconds. |
|
684 # |
|
685 when isMainModule: |
|
686 let expected = 8 |
|
687 var |
|
688 socket = newSocket() |
|
689 messages: seq[ StompResponse ] = @[] |
|
690 |
|
691 if paramCount() != 3: quit "See source comments for how to run functional tests." |
|
692 |
|
693 var stomp = newStompClient( socket, paramStr(2) ) |
|
694 stomp.connect |
|
695 echo stomp |
|
696 |
|
697 case paramStr(1): |
|
698 |
|
699 of "benchmark": |
|
700 echo "* Running for 10 seconds. Compile with -d:debug to see the Stomp conversation." |
|
701 var count = 0 |
|
702 var start = get_time() |
|
703 |
|
704 proc incr( c: StompClient, r: StompResponse ) = |
|
705 let id = r["ack"] |
|
706 count = count + 1 |
|
707 c.nack( id ) |
|
708 |
|
709 stomp.message_callback = incr |
|
710 stomp.subscribe( paramStr(3), "client" ) |
|
711 stomp.send( paramStr(3), "hi." ) |
|
712 while get_time() - start < 10: |
|
713 stomp.wait_for_messages( false ) |
|
714 |
|
715 printf "* Processed %d messages in 10 seconds.\n", count |
|
716 stomp.disconnect |
|
717 |
|
718 |
|
719 # Store incoming messages, ensure their contents match our expected behavior. |
|
720 # |
|
721 of "receiver": |
|
722 var heartbeats = 0 |
|
723 echo "* Waiting on messages from publisher. Compile with -d:debug to see the Stomp conversation." |
|
724 |
|
725 proc receive_message( c: StompClient, r: StompResponse ) = |
|
726 messages.add( r ) |
|
727 case r.frame: |
|
728 of "RECEIPT": |
|
729 discard |
|
730 of "MESSAGE": |
|
731 let body = r.payload |
|
732 let id = r[ "ack" ] |
|
733 |
|
734 proc seen_heartbeat( c: StompClient, r: StompResponse ) = |
|
735 heartbeats = heartbeats + 1 |
|
736 |
|
737 stomp.message_callback = receive_message |
|
738 stomp.receipt_callback = receive_message |
|
739 stomp.heartbeat_callback = seen_heartbeat |
|
740 stomp.subscribe( paramStr(3) ) |
|
741 |
|
742 # Populate the messages sequence with the count of expected messages. |
|
743 for i in 1..expected: stomp.wait_for_messages( false ) |
|
744 |
|
745 # Assertions on the results! |
|
746 # |
|
747 doAssert( messages.len == expected ) |
|
748 doAssert( messages[0].payload == nil ) |
|
749 |
|
750 doAssert( messages[1].payload == "Hello world!" ) |
|
751 |
|
752 doAssert( messages[2].payload == "Dumb.\n\n" ) |
|
753 |
|
754 doAssert( messages[3].payload == "Hello again." ) |
|
755 doAssert( messages[3][ "content-type" ] == "text/plain" ) |
|
756 doAssert( messages[3][ "Content-Type" ] == "text/plain" ) |
|
757 |
|
758 doAssert( messages[4][ "x-custom" ] == "yum" ) |
|
759 |
|
760 doAssert( messages[5][ "receipt" ] == "42" ) |
|
761 |
|
762 doAssert( messages[6].payload == "transaction!" ) |
|
763 doAssert( messages[7].payload == "transaction 2" ) |
|
764 |
|
765 stomp.disconnect |
|
766 |
|
767 if heartbeats > 0: |
|
768 printf "* Tests passed! %d heartbeats seen.", heartbeats |
|
769 else: |
|
770 echo "* Tests passed!" |
|
771 |
|
772 |
|
773 # Publish a variety of messages with various options. |
|
774 # Pause momentarily between sends(), as brokers -might- impose |
|
775 # rate limits and/or message dropping. |
|
776 # |
|
777 of "publisher": |
|
778 echo "* Publishing to receiver. Compile with -d:debug to see the Stomp conversation." |
|
779 |
|
780 # Simple, no frills event. |
|
781 stomp.send( paramStr(3) ) |
|
782 sleep 500 |
|
783 |
|
784 # Event with a body. |
|
785 stomp.send( paramStr(3), "Hello world!" ) |
|
786 sleep 500 |
|
787 |
|
788 # Event that doesn't contain a content-length. |
|
789 # (Note, the broker may elect to add one on your behalf, which is a good thing... |
|
790 # but invalidates this test.) |
|
791 stomp.socksend( "SEND" & CRLF ) |
|
792 stomp.socksend( "destination:" & paramStr(3) & CRLF & CRLF ) |
|
793 stomp.socksend( "Dumb.\n\n" & NULL ) |
|
794 sleep 500 |
|
795 |
|
796 # Content-Type |
|
797 stomp.send( paramStr(3), "Hello again.", "text/plain" ) |
|
798 sleep 500 |
|
799 |
|
800 # Custom headers. |
|
801 var headers: seq[ tuple[ name: string, value: string ] ] = @[] |
|
802 headers.add( ("x-custom", "yum") ) |
|
803 stomp.send( paramStr(3), "Hello again.", "text/plain", headers ) |
|
804 sleep 500 |
|
805 |
|
806 # Receipt requests. |
|
807 proc receive_receipt( c: StompClient, r: StompResponse ) = |
|
808 messages.add( r ) |
|
809 headers = @[] |
|
810 headers.add( ("receipt", "42") ) |
|
811 stomp.send( paramStr(3), "Hello again.", "text/plain", headers ) |
|
812 stomp.receipt_callback = receive_receipt |
|
813 stomp.wait_for_messages( false ) |
|
814 doAssert( messages[0]["receipt-id"] == "42" ) |
|
815 |
|
816 # Aborted transaction. |
|
817 stomp.begin( "test-abort" ) |
|
818 for i in 1..3: |
|
819 stomp.send( paramStr(3), "Message: " & $i ) |
|
820 stomp.abort |
|
821 |
|
822 # Committed transaction. |
|
823 stomp.begin( "test-commit" ) |
|
824 stomp.send( paramStr(3), "transaction!" ) |
|
825 stomp.commit |
|
826 |
|
827 # Mixed transactions. |
|
828 for i in 1..3: |
|
829 headers = @[] |
|
830 headers.add( ("transaction", "test-" & $i ) ) |
|
831 stomp.begin( "test-" & $i ) |
|
832 stomp.send( paramStr(3), "transaction " & $i, nil, headers ) |
|
833 sleep 500 |
|
834 stomp.abort( "test-1" ) |
|
835 sleep 500 |
|
836 stomp.commit( "test-2" ) |
|
837 sleep 500 |
|
838 stomp.abort( "test-3" ) |
|
839 sleep 500 |
|
840 |
|
841 stomp.disconnect |
|
842 echo "* Tests passed!" |
|
843 |
|
844 else: |
|
845 quit "See source comments for how to run functional tests." |
|
846 |
|
847 |
|