1 # -*- encoding: utf-8 -*-
3 if Kernel.respond_to?(:require_relative)
4 require_relative("test_helper")
6 $:.unshift(File.dirname(__FILE__))
12 Main class for testing Stomp::Client instances.
15 class TestClient < Test::Unit::TestCase
19 @client = get_client()
20 # Multi_thread test data
23 @tcldbg = ENV['TCLDBG'] || ENV['TDBGALL'] ? true : false
27 @client.close if @client && @client.open? # allow tests to close
32 mn = "test_poll_async" if @tcldbg
33 p [ "01", mn, "starts" ] if @tcldbg
34 # If the test 'hangs' here, Connection#poll is broken.
37 p [ "99", mn, "ends" ] if @tcldbg
38 end unless RUBY_ENGINE =~ /jruby/
41 def test_ack_api_works
42 mn = "test_ack_api_works" if @tcldbg
43 p [ "01", mn, "starts" ] if @tcldbg
44 @client.publish make_destination, message_text, {:suppress_content_length => true}
47 @client.subscribe(make_destination, {:ack => 'client'}) {|msg|
49 p [ "02", mn, "have_msg" ] if @tcldbg
51 p [ "03", mn, "sub_done" ] if @tcldbg
52 sleep 0.01 until received
53 assert_equal message_text, received.body
56 if @client.protocol == Stomp::SPL_11 # 1.1 only
57 ack_headers["subscription"] = received.headers["subscription"]
59 @client.acknowledge(received, ack_headers) {|r|
61 p [ "04", mn, "have_rcpt" ] if @tcldbg
63 p [ "05", mn, "ack_sent" ] if @tcldbg
64 sleep 0.01 until receipt
65 assert_not_nil receipt.headers['receipt-id']
66 checkEmsg(@client) unless jruby?()
67 p [ "99", mn, "ends" ] if @tcldbg
70 # Test Client subscribe
71 def test_asynch_subscribe
72 mn = "test_async_subscribe" if @tcldbg
73 p [ "01", mn, "starts" ] if @tcldbg
75 @client.subscribe(make_destination) {|msg| received = msg}
76 @client.publish make_destination, message_text
77 sleep 0.01 until received
79 assert_equal message_text, received.body
80 checkEmsg(@client) unless jruby?()
81 p [ "99", mn, "ends" ] if @tcldbg
84 # Test not ACKing messages.
86 mn = "test_noack" if @tcldbg
87 p [ "01", mn, "starts" ] if @tcldbg
89 @client.publish make_destination, message_text
92 @client.subscribe(make_destination, :ack => :client) {|msg| received = msg}
93 sleep 0.01 until received
94 assert_equal message_text, received.body
97 # was never acked so should be resent to next client
99 @client = get_client()
101 @client.subscribe(make_destination) {|msg| received2 = msg}
102 sleep 0.01 until received2
104 assert_equal message_text, received2.body
105 assert_equal received.body, received2.body
106 assert_equal received.headers['message-id'], received2.headers['message-id'] unless ENV['STOMP_RABBIT']
108 p [ "99", mn, "ends" ] if @tcldbg
109 end unless RUBY_ENGINE =~ /jruby/
111 # Test obtaining a RECEIPT via a listener.
113 mn = "test_receipts" if @tcldbg
114 p [ "01", mn, "starts" ] if @tcldbg
117 @client.publish(make_destination, message_text) {|r| receipt = r}
118 sleep 0.1 until receipt
119 assert_equal receipt.command, Stomp::CMD_RECEIPT
121 @client.subscribe(make_destination) {|m| message = m}
122 sleep 0.1 until message
123 assert_equal message_text, message.body
124 checkEmsg(@client) unless jruby?()
125 p [ "99", mn, "ends" ] if @tcldbg
128 # Test requesting a receipt on disconnect.
129 def test_disconnect_receipt
130 mn = "test_disconnect_receipt" if @tcldbg
131 p [ "01", mn, "starts" ] if @tcldbg
133 @client.close :receipt => "xyz789"
134 assert_not_nil(@client.disconnect_receipt, "should have a receipt")
135 assert_equal(@client.disconnect_receipt.headers['receipt-id'],
136 "xyz789", "receipt sent and received should match")
137 p [ "99", mn, "ends" ] if @tcldbg
140 # Test publish and immediate subscribe.
141 def test_publish_then_sub
142 mn = "test_publish_then_sub" if @tcldbg
143 p [ "01", mn, "starts" ] if @tcldbg
145 @client.publish make_destination, message_text
147 @client.subscribe(make_destination) {|m| message = m}
148 sleep 0.01 until message
150 assert_equal message_text, message.body
151 checkEmsg(@client) unless jruby?()
152 p [ "99", mn, "ends" ] if @tcldbg
155 # Test that Client subscribe requires a block.
156 def test_subscribe_requires_block
157 mn = "test_subscribe_requires_block" if @tcldbg
158 p [ "01", mn, "starts" ] if @tcldbg
160 assert_raise(Stomp::Error::NoListenerGiven) do
161 @client.subscribe make_destination
163 checkEmsg(@client) unless jruby?()
164 p [ "99", mn, "ends" ] if @tcldbg
167 # Test transaction publish.
168 def test_transactional_publish
169 mn = "test_transactional_publish" if @tcldbg
170 p [ "01", mn, "starts" ] if @tcldbg
174 @client.publish make_destination, message_text, :transaction => tid
178 @client.subscribe(make_destination) {|m| message = m}
179 sleep 0.01 until message
181 assert_equal message_text, message.body
182 checkEmsg(@client) unless jruby?()
183 p [ "99", mn, "ends" ] if @tcldbg
184 end unless ENV['STOMP_ARTEMIS']
186 # Test transaction publish and abort.
187 def test_transaction_publish_then_rollback
188 mn = "test_transaction_publish_then_rollback" if @tcldbg
189 p [ "01", mn, "starts" ] if @tcldbg
193 @client.publish make_destination, "first_message", :transaction => tid
197 @client.publish make_destination, "second_message", :transaction => tid
201 @client.subscribe(make_destination) {|m| message = m}
202 sleep 0.01 until message
203 assert_equal "second_message", message.body
204 checkEmsg(@client) unless jruby?()
205 p [ "99", mn, "ends" ] if @tcldbg
206 end unless ENV['STOMP_ARTEMIS']
208 # Test transaction publish and abort, receive with new client.
209 # New client uses ack => client.
210 def test_tran_ack_abrt_newcli_cli
211 mn = "test_tran_ack_abrt_newcli_cli" if @tcldbg
212 p [ "01", mn, "starts" ] if @tcldbg
215 @client.close if @client && @client.open? # allow tests to close
216 @client = get_client()
219 @client.publish q, data
224 if @client.protocol() == Stomp::SPL_10
225 @client.subscribe(q, :ack => 'client') {|m| message = m}
226 else # 1.1 and 1.2 are the same for this
228 @client.subscribe(q, :ack => 'client', :id => sid) {|m| message = m}
230 sleep 0.01 until message
231 assert_equal data, message.body
232 case @client.protocol()
234 @client.acknowledge message, :transaction => tid
235 checkEmsg(@client) unless jruby?()
237 @client.acknowledge message, :transaction => tid, :subscription => message.headers['subscription']
238 checkEmsg(@client) unless jruby?()
240 @client.acknowledge message, :transaction => tid, :id => message.headers['ack']
241 checkEmsg(@client) unless jruby?()
243 message = nil # reset
244 @client.abort tid # now abort
245 checkEmsg(@client) unless jruby?()
246 # lets recreate the connection
248 @client = get_client()
253 if @client.protocol() == Stomp::SPL_10
254 @client.subscribe(q, :ack => 'client') {|m| message2 = m}
255 else # 1.1 and 1.2 are the same for this
257 @client.subscribe(q, :ack => 'client', :id => sid) {|m| message2 = m}
259 sleep 0.01 until message2
260 assert_not_nil message
261 assert_equal data, message2.body
262 case @client.protocol()
264 @client.acknowledge message2, :transaction => tid2
265 checkEmsg(@client) unless jruby?()
267 @client.acknowledge message2, :transaction => tid2, :subscription => message2.headers['subscription']
268 checkEmsg(@client) unless jruby?()
270 @client.acknowledge message2, :transaction => tid2, :id => message2.headers['ack']
271 checkEmsg(@client) unless jruby?()
274 checkEmsg(@client) unless jruby?()
276 p [ "99", mn, "ends" ] if @tcldbg
277 end unless ENV['STOMP_ARTEMIS'] # See Artemis docs for 1.3, page 222
279 # Test transaction publish and abort, receive with new client.
280 # New client uses ack => auto.
281 def test_tran_ack_abrt_newcli_auto
282 mn = "test_tran_ack_abrt_newcli_auto" if @tcldbg
283 p [ "01", mn, "starts" ] if @tcldbg
286 @client.close if @client && @client.open? # allow tests to close
287 @client = get_client()
290 @client.publish q, data
295 if @client.protocol() == Stomp::SPL_10
296 @client.subscribe(q, :ack => 'client') {|m| message = m}
297 else # 1.1 and 1.2 are the same for this
299 @client.subscribe(q, :ack => 'client', :id => sid) {|m| message = m}
301 sleep 0.01 until message
302 assert_equal data, message.body
303 case @client.protocol()
305 @client.acknowledge message, :transaction => tid
306 checkEmsg(@client) unless jruby?()
308 @client.acknowledge message, :transaction => tid, :subscription => message.headers['subscription']
309 checkEmsg(@client) unless jruby?()
311 @client.acknowledge message, :transaction => tid, :id => message.headers['ack']
312 checkEmsg(@client) unless jruby?()
314 message = nil # reset
315 @client.abort tid # now abort
316 checkEmsg(@client) unless jruby?()
317 # lets recreate the connection
320 @client = get_client()
325 if @client.protocol() == Stomp::SPL_10
326 @client.subscribe(q, :ack => 'auto') {|m| message2 = m}
327 else # 1.1 and 1.2 are the same for this
329 @client.subscribe(q, :ack => 'auto', :id => sid) {|m| message2 = m}
331 sleep 0.01 until message2
332 assert_not_nil message2
333 assert_equal data, message2.body
335 checkEmsg(@client) unless jruby?()
337 p [ "99", mn, "ends" ] if @tcldbg
338 end unless ENV['STOMP_ARTEMIS'] # See Artemis docs for 1.3, page 222
340 # Test that subscription destinations must be unique for a Client.
341 def test_raise_on_multiple_subscriptions_to_same_make_destination
342 mn = "test_raise_on_multiple_subscriptions_to_same_make_destination" if @tcldbg
343 p [ "01", mn, "starts" ] if @tcldbg
345 subscribe_dest = make_destination
346 @client.subscribe(subscribe_dest) {|m| nil }
347 assert_raise(Stomp::Error::DuplicateSubscription) do
348 @client.subscribe(subscribe_dest) {|m| nil }
350 checkEmsg(@client) unless jruby?()
351 p [ "99", mn, "ends" ] if @tcldbg
354 # Test that subscription IDs must be unique for a Client.
355 def test_raise_on_multiple_subscriptions_to_same_id
356 mn = "test_raise_on_multiple_subscriptions_to_same_id" if @tcldbg
357 p [ "01", mn, "starts" ] if @tcldbg
359 subscribe_dest = make_destination
360 @client.subscribe(subscribe_dest, {'id' => 'myid'}) {|m| nil }
361 assert_raise(Stomp::Error::DuplicateSubscription) do
362 @client.subscribe(subscribe_dest, {'id' => 'myid'}) {|m| nil }
364 checkEmsg(@client) unless jruby?()
365 p [ "99", mn, "ends" ] if @tcldbg
368 # Test that subscription IDs must be unique for a Client, mixed id specification.
369 def test_raise_on_multiple_subscriptions_to_same_id_mixed
370 mn = "test_raise_on_multiple_subscriptions_to_same_id_mixed" if @tcldbg
371 p [ "01", mn, "starts" ] if @tcldbg
373 subscribe_dest = make_destination
374 @client.subscribe(subscribe_dest, {'id' => 'myid'}) {|m| nil }
375 assert_raise(Stomp::Error::DuplicateSubscription) do
376 @client.subscribe(subscribe_dest, {:id => 'myid'}) {|m| nil }
378 checkEmsg(@client) unless jruby?()
379 p [ "99", mn, "ends" ] if @tcldbg
382 # Test wildcard subscribe. Primarily for AMQ.
383 def test_asterisk_wildcard_subscribe
384 mn = "test_asterisk_wildcard_subscribe" if @tcldbg
385 p [ "01", mn, "starts" ] if @tcldbg
387 queue_base_name = make_destination
388 queue1 = queue_base_name + ".a"
389 queue2 = queue_base_name + ".b"
390 send_message = message_text
391 @client.publish queue1, send_message
392 @client.publish queue2, send_message
394 @client.subscribe(queue_base_name + ".*", :ack => 'client') do |m|
396 @client.acknowledge(m)
398 Timeout::timeout(4) do
399 sleep 0.1 while messages.size < 2
402 messages.each do |message|
403 assert_not_nil message
404 assert_equal send_message, message.body
406 results = [queue1, queue2].collect do |queue|
407 messages.any? do |message|
408 message_source = message.headers['destination']
409 message_source == queue
412 assert results.all?{|a| a == true }
414 p [ "99", mn, "ends" ] if @tcldbg
415 end unless ENV['STOMP_NOWILD']
417 # Test wildcard subscribe with >. Primarily for AMQ.
418 def test_greater_than_wildcard_subscribe
419 mn = "test_greater_than_wildcard_subscribe" if @tcldbg
420 p [ "01", mn, "starts" ] if @tcldbg
422 queue_base_name = make_destination + "."
423 queue1 = queue_base_name + "foo.a"
424 queue2 = queue_base_name + "bar.a"
425 queue3 = queue_base_name + "foo.b"
426 send_message = message_text
427 @client.publish queue1, send_message
428 @client.publish queue2, send_message
429 @client.publish queue3, send_message
431 # should subscribe to all three queues
432 @client.subscribe(queue_base_name + ">", :ack => 'client') do |m|
434 @client.acknowledge(m)
436 Timeout::timeout(4) do
437 sleep 0.1 while messages.size < 3
440 messages.each do |message|
441 assert_not_nil message
442 assert_equal send_message, message.body
444 # make sure that the messages received came from the expected queues
445 results = [queue1, queue2, queue3].collect do |queue|
446 messages.any? do |message|
447 message_source = message.headers['destination']
448 message_source == queue
451 assert results.all?{|a| a == true }
453 p [ "99", mn, "ends" ] if @tcldbg
454 end unless ENV['STOMP_NOWILD'] || ENV['STOMP_DOTQUEUE']
456 # Test transaction with client side reacknowledge.
457 def test_transaction_with_client_side_reack
458 mn = "test_transaction_with_client_side_reack" if @tcldbg
459 p [ "01", mn, "starts" ] if @tcldbg
461 @client.close if @client && @client.open? # allow tests to close
462 @client = get_client()
465 @client.publish q, data
470 if @client.protocol() == Stomp::SPL_10
471 @client.subscribe(q, :ack => 'client') { |m| message = m }
474 @client.subscribe(q, :ack => 'client', :id => sid) { |m| message = m }
476 sleep 0.1 while message.nil?
477 assert_equal data, message.body
478 case @client.protocol()
480 @client.acknowledge message, :transaction => tid
481 checkEmsg(@client) unless jruby?()
483 @client.acknowledge message, :transaction => tid, :subscription => message.headers['subscription']
484 checkEmsg(@client) unless jruby?()
486 @client.acknowledge message, :transaction => tid, :id => message.headers['ack']
487 checkEmsg(@client) unless jruby?()
491 # Wait for redlivery (Client logic)
492 sleep 0.1 while message.nil?
493 assert_not_nil message
494 assert_equal data, message.body
497 case @client.protocol()
499 @client.acknowledge message, :transaction => tid2
500 checkEmsg(@client) unless jruby?()
502 @client.acknowledge message, :transaction => tid2, :subscription => message.headers['subscription']
503 checkEmsg(@client) unless jruby?()
505 @client.acknowledge message, :transaction => tid2, :id => message.headers['ack']
506 checkEmsg(@client) unless jruby?()
509 checkEmsg(@client) unless jruby?()
512 p [ "99", mn, "ends" ] if @tcldbg
513 end unless ENV['STOMP_ARTEMIS']
515 # Test that a connection frame is received.
516 def test_connection_frame
517 mn = "test_connection_frame" if @tcldbg
518 p [ "01", mn, "starts" ] if @tcldbg
520 assert_not_nil @client.connection_frame
522 p [ "99", mn, "ends" ] if @tcldbg
523 end unless RUBY_ENGINE =~ /jruby/
525 # Test basic unsubscribe.
527 mn = "test_unsubscribe" if @tcldbg
528 p [ "01", mn, "starts" ] if @tcldbg
530 @client.close if @client && @client.open? # close setup work
533 dest = make_destination
534 to_send = message_text
535 client = get_client()
537 if client.protocol() == Stomp::SPL_10
538 client.subscribe(dest, :ack => 'client') { |m| message = m }
541 client.subscribe(dest, :ack => 'client', :id => sid) { |m| message = m }
543 client.publish dest, to_send
544 Timeout::timeout(4) do
545 sleep 0.01 until message
547 assert_equal to_send, message.body, "first body check"
548 if client.protocol() == Stomp::SPL_10
549 client.unsubscribe dest
551 client.unsubscribe dest, :id => sid
554 # Same message should remain on the queue. Receive it again with ack=>auto.
556 client = get_client()
557 if client.protocol() == Stomp::SPL_10
558 client.subscribe(dest, :ack => 'auto') { |m| message_copy = m }
561 client.subscribe(dest, :ack => 'auto', :id => sid) { |m| message_copy = m }
563 Timeout::timeout(4) do
564 sleep 0.01 until message_copy
566 assert_equal to_send, message_copy.body, "second body check"
567 assert_equal message.headers['message-id'], message_copy.headers['message-id'], "header check" unless ENV['STOMP_RABBIT']
568 checkEmsg(client) unless jruby?()
570 p [ "99", mn, "ends" ] if @tcldbg
573 # Test subscribe from a worker thread.
574 def test_thread_one_subscribe
575 mn = "test_thread_one_subscribe" if @tcldbg
576 p [ "01", mn, "starts" ] if @tcldbg
579 dest = make_destination
580 Thread.new(@client) do |acli|
582 if acli.protocol() == Stomp::SPL_10
583 acli.subscribe(dest) { |m| msg = m }
585 acli.subscribe(dest, :id => acli.uuid()) { |m| msg = m }
587 Timeout::timeout(4) do
592 @client.publish(dest, message_text)
596 p [ "99", mn, "ends" ] if @tcldbg
597 end unless RUBY_ENGINE =~ /jruby/
599 # Test subscribe from multiple worker threads.
600 def test_thread_multi_subscribe
601 mn = "test_thread_multi_subscribe" if @tcldbg
602 p [ "01", mn, "starts" ] if @tcldbg
607 dest = make_destination
608 1.upto(@max_threads) do |tnum|
609 # Threads within threads .....
610 Thread.new(@client) do |acli|
613 if acli.protocol() == Stomp::SPL_10
614 acli.subscribe(dest) { |m|
619 # Simulate message processing
623 acli.subscribe(dest, :id => acli.uuid()) { |m|
628 # Simulate message processing
635 1.upto(@max_msgs) do |mnum|
636 msg = Time.now.to_s + " #{mnum}"
637 @client.publish(dest, msg)
640 max_sleep = (RUBY_VERSION =~ /1\.8\.6/) ? 30 : 5
644 break if @max_msgs == msg_ctr
645 total_slept += sleep_incr
646 break if total_slept > max_sleep
649 assert_equal @max_msgs, msg_ctr
650 checkEmsg(@client) unless jruby?()
651 p [ "99", mn, "ends" ] if @tcldbg
654 # Test that methods detect no client connection is present.
655 def test_closed_checks_client
656 mn = "test_closed_checks_client" if @tcldbg
657 p [ "01", mn, "starts" ] if @tcldbg
661 assert_raise Stomp::Error::NoCurrentConnection do
662 m = Stomp::Message.new("")
663 @client.acknowledge(m) {|r| _ = r}
666 assert_raise Stomp::Error::NoCurrentConnection do
667 @client.begin("dummy_data")
670 assert_raise Stomp::Error::NoCurrentConnection do
671 @client.commit("dummy_data")
674 assert_raise Stomp::Error::NoCurrentConnection do
675 @client.abort("dummy_data")
678 assert_raise Stomp::Error::NoCurrentConnection do
679 @client.subscribe("dummy_data", {:ack => 'auto'}) {|msg| _ = msg}
682 assert_raise Stomp::Error::NoCurrentConnection do
683 @client.unsubscribe("dummy_data")
686 assert_raise Stomp::Error::NoCurrentConnection do
687 @client.publish("dummy_data","dummy_data")
690 assert_raise Stomp::Error::NoCurrentConnection do
691 @client.unreceive("dummy_data")
694 assert_raise Stomp::Error::NoCurrentConnection do
695 @client.close("dummy_data")
697 p [ "99", mn, "ends" ] if @tcldbg
700 # test JRuby detection
701 def test_jruby_presence
702 mn = "test_jruby_presence" if @tcldbg
703 p [ "01", mn, "starts" ] if @tcldbg
705 if defined?(RUBY_ENGINE) && RUBY_ENGINE =~ /jruby/
706 assert @client.jruby?
708 assert !@client.jruby?
710 p [ "99", mn, "ends" ] if @tcldbg
713 # test max redeliveries is not broken (6c2c1c1)
714 def test_max_redeliveries
715 mn = "test_max_redeliveries" if @tcldbg
716 p [ "01", mn, "starts" ] if @tcldbg
719 2.upto(2) do |max_re|
720 rdmsg = "To Be Redelivered: #{max_re.to_s}"
721 dest = make_destination + ".#{max_re.to_s}"
722 @client = get_client()
723 sid = "sid_max_redeliveries_#{max_re.to_s}"
726 sh = @client.protocol() == Stomp::SPL_10 ? {} : {:id => sid}
727 @client.subscribe(dest, sh) {|msg|
729 @client.unreceive(msg, :max_redeliveries => max_re,
730 :dead_letter_queue => make_dlq())
731 received = msg if rm_actual - 1 == max_re
733 @client.publish(dest, rdmsg)
734 sleep 0.01 until received
735 assert_equal rdmsg, received.body
737 @client.unsubscribe dest, sh
738 assert_equal max_re, rm_actual - 1
741 p [ "99", mn, "ends" ] if @tcldbg
742 end unless ENV['STOMP_ARTEMIS'] # need to investigate this, but skip
745 # test issue99, OK values
746 def test_cli_iss99_ok
748 return unless host() == "localhost" && port() == 61613
749 mn = "test_cli_iss99_ok" if @tcldbg
750 p [ "01", mn, "starts" ] if @tcldbg
754 ok_vals = dflt_data_ok()
755 ok_vals.each do |hsv|
756 cli = Stomp::Client.open(hsv)
759 p [ "99", mn, "ends" ] if @tcldbg
762 # test issue99, exception values
763 def test_cli_iss99_ex
764 return unless host() == "localhost" && port() == 61613
765 mn = "test_cli_iss99_ex" if @tcldbg
766 p [ "01", mn, "starts" ] if @tcldbg
770 ex_vals = dflt_data_ex()
771 ex_vals.each do |hsv|
772 assert_raise ArgumentError do
773 _ = Stomp::Client.open(hsv)
776 p [ "99", mn, "ends" ] if @tcldbg
779 def test_cli_nodest_sub
780 mn = "test_cli_nodest_sub" if @tcldbg
781 p [ "01", mn, "starts" ] if @tcldbg
783 assert_raise Stomp::Error::DestinationRequired do
784 @client.subscribe(nil) {|msg| puts msg}
786 p [ "99", mn, "ends" ] if @tcldbg
789 def test_cli_nodest_unsub
790 mn = "test_cli_nodest_unsub" if @tcldbg
791 p [ "01", mn, "starts" ] if @tcldbg
793 assert_raise Stomp::Error::DestinationRequired do
794 @client.unsubscribe(nil)
796 p [ "99", mn, "ends" ] if @tcldbg
799 def test_cli_nodest_pub
800 mn = "test_cli_nodest_pub" if @tcldbg
801 p [ "01", mn, "starts" ] if @tcldbg
803 assert_raise Stomp::Error::DestinationRequired do
804 @client.publish(nil, "msg")
806 p [ "99", mn, "ends" ] if @tcldbg
811 name = caller_method_name unless name
812 "test_client#" + name