PROTON-781: Reactive Ruby examples
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/f63454c5 Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/f63454c5 Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/f63454c5 Branch: refs/heads/PROTON-781-ruby-reactor-apis Commit: f63454c5caac0124f600793a96d9c560c15fa22f Parents: 81c5cbd Author: Darryl L. Pierce <mcpie...@gmail.com> Authored: Tue May 26 15:34:11 2015 -0400 Committer: Darryl L. Pierce <mcpie...@gmail.com> Committed: Mon Jun 8 13:59:49 2015 -0400 ---------------------------------------------------------------------- examples/ruby/lib/debugging.rb | 26 +++ examples/ruby/lib/send_and_receive.rb | 90 +++++++++++ examples/ruby/reactor/README.md | 103 ++++++++++++ examples/ruby/reactor/broker.rb | 200 ++++++++++++++++++++++++ examples/ruby/reactor/client.rb | 68 ++++++++ examples/ruby/reactor/direct_recv.rb | 60 +++++++ examples/ruby/reactor/direct_send.rb | 59 +++++++ examples/ruby/reactor/helloworld.rb | 69 ++++++++ examples/ruby/reactor/helloworld_direct.rb | 74 +++++++++ examples/ruby/reactor/server.rb | 64 ++++++++ examples/ruby/reactor/simple_recv.rb | 58 +++++++ examples/ruby/reactor/simple_send.rb | 55 +++++++ 12 files changed, 926 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/lib/debugging.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/lib/debugging.rb b/examples/ruby/lib/debugging.rb new file mode 100644 index 0000000..5065d51 --- /dev/null +++ b/examples/ruby/lib/debugging.rb @@ -0,0 +1,26 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +module Debugging + + def debug(text) + print "[#{Time.now.strftime('%s')}] #{text}\n" + end + +end http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/lib/send_and_receive.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/lib/send_and_receive.rb b/examples/ruby/lib/send_and_receive.rb new file mode 100644 index 0000000..9fd7417 --- /dev/null +++ b/examples/ruby/lib/send_and_receive.rb @@ -0,0 +1,90 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +class ExampleSend < Qpid::Proton::Handler::MessagingHandler + + attr_reader :url + + def initialize(url, expected) + super() + @url = url + @sent = 0 + @confirmed = 0 + @expected = expected + end + + def on_sendable(event) + while event.sender.credit > 0 && @sent < @expected + msg = Qpid::Proton::Message.new + msg.body = "sequence #{@sent}" + msg.id = @sent + event.sender.send(msg) + @sent = @sent + 1 + end + end + + def on_accepted(event) + @confirmed = @confirmed + 1 + if self.finished? + puts "#{@expected > 1 ? 'All ' : ''}#{@expected} message#{@expected > 1 ? 's' : ''} confirmed!" + event.connection.close + end + end + + def on_disconnected(event) + @sent = @confirmed + end + + def finished? + @confirmed == @expected + end + +end + +class ExampleReceive < Qpid::Proton::Handler::MessagingHandler + + attr_reader :url + + def initialize(url, expected) + super() + @url = url + @expected = expected + @received = 0 + end + + def on_message(event) + if event.message.id.nil? || event.message.id < @received + puts "Missing or old message id: id=#{event.message.id}" + return + end + if @expected.zero? || (@received < @expected) + puts "Received: #{event.message.body}" + @received = @received + 1 + if finished? + event.receiver.close + event.connection.close + end + end + end + + def finished? + @received == @expected + end + +end http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/reactor/README.md ---------------------------------------------------------------------- diff --git a/examples/ruby/reactor/README.md b/examples/ruby/reactor/README.md new file mode 100644 index 0000000..38cc6ba --- /dev/null +++ b/examples/ruby/reactor/README.md @@ -0,0 +1,103 @@ +## What Is The Reactor? + +A little outside of the scope of this document, but the reactor is an event source for letting an application know about events in the Proton messaging system. With this set of APIs an application can be register handlers that are notified when a connection is created, a message received, or a session closes. + +### Handlers + +An application creates **handlers**, objects which provide methods through which the reactor notifies the application's components of events and allows them each to handle the ones in which they are interested (see the Chain Of Responsibility design pattern for more on this idea). There are some pre-defined handlers for responding to incoming message events, outgoing message events, data flow and managing the AMQP endpoints. Look in the **Qpid::Proton::Handlers** package for more details on these classes. + +## Simple Reactor Examples + +### The Broker + +The reactor examples come with a sample broker which can be used by other examples and which also works as an example itself. For now we'll just start up the broker example and tell it to listen on port 8888: + +```` +$ ruby ../examples/ruby/reactor/broker.rb --address=0.0.0.0:8888 +Listening on 0.0.0.0:8888 +```` + +This example broker will receive messages, create queues as needed, and deliver messages to endpoints. + +### Hello World Using A Broker + +Our first example creates an endpoint that sends messages to a queue to which it is subscribed. So it both sends and receives its message in one pass. + +To start it, simply run: + +``` +$ ruby ../examples/ruby/reactor/helloworld.rb --address=0.0.0.0:8888 --queue=examples +Hello world! +``` + +As you can see, the classic message was output by the example. Now let's take a look at what's going on under the covers. + +#### Events When Talking To A Broker + +The following events occur while **helloworld.rb** runs: + + * **on_start** - Fired when the application is started. + * **on_sendable** - Fired when a message can be sent. + * **on_message** - Fired when a message is received. + +### Hello World Without A Broker required + +The next example we'll look at will send the classic "Hello world" message to itself directly. This example shows some very fundamental elements of the reactor APIs that you should understand. + +To launch the example: + +``` + $ ruby helloworld_direct.rb --address=0.0.0.0:8888/examples + Hello world! +``` + +Not very different from the example that uses the broker, which is what we'd expect from the outside. But let's take a look inside of the example and see how it's different at that level + +The direct version takes on the responsibility for listening to incoming connections as well as making an outgoing connection. So we see the following additional events occurring: + + * **on_accepted** - Fired when a message is received. + * **on_connection_closed** - Fired when an endpoint closes its connection. + +## More Complex Reactor Examples + +Now that we've covered the basics with the archetypical hello world app, let's look at some more interesting examples. + +There are four example applications that demonstrate how to send and receive messages both directly and through an intermediary, such as a broker: + + * **simple_send.rb** - sends messages to a receiver at a specific address and receives responses via an intermediary, + * **simple_recv.rb** - receives messages from via an intermediary, + * **direct_send.rb** - sends messages directly to a receiver and listens for responses itself, and + * **direct_recv.rb** - receives messages directly. + + Simple send and direct send may, at first, seem to be so similar that you wonder why they're not just the same applciation. And I know for me I was wonder as I wrote the list above why there were two examples. The reason is that **simple_send.rb** uses the intermediary transfer responses to the messages it sends, while **direct_send.rb** uses an *Acceptor* to listen for an process responses. + + You can use the examples in the follow ways: + + ``` + simple_send.rb -> broker <- simple_recv.rb + simple_send.rb -> direct_recv.rb + direct_send.rb -> simple_recv.rb + ``` + +In this set of examples we see the following event occurring, in addition to what we've seen before: + + * **on_disconnected** - Fired when the transport is closed. + +## Now About That Broker example + +The **broker.rb** example application is a nice demonstration of doing something more interesting in Ruby with Proton. + +The way the broker works is to listen to incoming connections, examine the components of the address for that connection, attach that connection to an exchange managing that address and then it sends any messages destined for that address to them. + +The components of the broker example include: + * **Broker** - A class that extends the MessagingHandler class. It accepts incoming connections, manages subscribing them to exchanges, and transfers messages between them. + * **Exchange** - A class that represents a message queue, tracking what endpoints are subscribed to it. + +The Broker manages a map connecting a queue address to the instance of Exchange that holds references to the endpoints of interest. + +The broker application demonstrates a new set of reactor events: + + * **on_link_opening** - Fired when a remote link is opened but the local end is not yet open. From this event the broker grabs the address and subscribes the link to an exchange for that address. + * **on_link_closing** - Fired when a remote link is closed but the local end is still open. From this event the broker grabs the address and unsubscribes the link from that exchange. + * **on_connection_closing** - Fired when a remote connection is closed but the local end is still open. + * **on_disconnected** - Fired when the protocol transport has closed. The broker removes all links for the disconnected connection, avoiding workign with endpoints that are now gone. http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/reactor/broker.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/reactor/broker.rb b/examples/ruby/reactor/broker.rb new file mode 100644 index 0000000..9d7e5be --- /dev/null +++ b/examples/ruby/reactor/broker.rb @@ -0,0 +1,200 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +require 'qpid_proton' +require 'optparse' +require 'pathname' + +require_relative '../lib/debugging' + +class Exchange + + include Debugging + + def initialize(dynamic = false) + @dynamic = dynamic + @queue = Queue.new + @consumers = [] + end + + def subscribe(consumer) + debug("subscribing #{consumer}") if $options[:debug] + @consumers << (consumer) + debug(" there are #{@consumers.size} consumers") if $options[:debug] + end + + def unsubscribe(consumer) + debug("unsubscribing #{consumer}") if $options[:debug] + if @consumers.include?(consumer) + @consumers.delete(consumer) + else + debug(" consumer doesn't exist") if $options[:debug] + end + debug(" there are #{@consumers.size} consumers") if $options[:debug] + @consumers.empty? && (@dynamic || @queue.empty?) + end + + def publish(message) + debug("queueing message: #{message.body}") if $options[:debug] + @queue << message + self.dispatch + end + + def dispatch(consumer = nil) + debug("dispatching: consumer=#{consumer}") if $options[:debug] + if consumer + c = [consumer] + else + c = @consumers + end + + while self.deliver_to(c) do + end + end + + def deliver_to(consumers) + debug("delivering to #{consumers.size} consumer(s)") if $options[:debug] + result = false + consumers.each do |consumer| + debug(" current consumer=#{consumer} credit=#{consumer.credit}") if $options[:debug] + if consumer.credit > 0 && !@queue.empty? + consumer.send(@queue.pop(true)) + result = true + end + end + return result + end + +end + +class Broker < Qpid::Proton::Handler::MessagingHandler + + include Debugging + + def initialize(url) + super() + @url = url + @queues = {} + end + + def on_start(event) + debug("on_start event") if $options[:debug] + @acceptor = event.container.listen(@url) + print "Listening on #{@url}\n" + end + + def queue(address) + debug("fetching queue for #{address}: (there are #{@queues.size} queues)") if $options[:debug] + unless @queues.has_key?(address) + debug(" creating new queue") if $options[:debug] + @queues[address] = Exchange.new + else + debug(" using existing queue") if $options[:debug] + end + result = @queues[address] + debug(" returning #{result}") if $options[:debug] + return result + end + + def on_link_opening(event) + debug("processing on_link_opening") if $options[:debug] + debug("link is#{event.link.sender? ? '' : ' not'} a sender") if $options[:debug] + if event.link.sender? + if event.link.remote_source.dynamic? + address = generate_uuid + event.link.source.address = address + q = Exchange.new(true) + @queues[address] = q + q.subscribe(event.link) + elsif event.link.remote_source.address + event.link.source.address = event.link.remote_source.address + self.queue(event.link.source.address).subscribe(event.link) + end + elsif event.link.remote_target.address + event.link.target.address = event.link.remote_target.address + end + end + + def unsubscribe(link) + debug("unsubscribing #{link.address}") if $options[:debug] + if @queues.has_key?(link.source.address) + if @queues[link.source.address].unsubscribe(link) + @queues.delete(link.source.address) + end + end + end + + def on_link_closing(event) + self.unsubscribe(event.link) if event.link.sender? + end + + def on_connection_closing(event) + self.remove_stale_consumers(event.connection) + end + + def on_disconnected(event) + self.remove_stale_consumers(event.connection) + end + + def remove_stale_consumers(connection) + l = connection.link_head(Qpid::Proton::Endpoint::REMOTE_ACTIVE) + while !l.nil? + self.unsubscribe(l) if l.sender? + l = l.next(Qpid::Proton::Endpoint::REMOTE_ACTIVE) + end + end + + def on_sendable(event) + debug("on_sendable event") if $options[:debug] + q = self.queue(event.link.source.address) + debug(" dispatching #{event.message} to #{q}") if $options[:debug] + q.dispatch(event.link) + end + + def on_message(event) + debug("on_message event") if $options[:debug] + q = self.queue(event.link.target.address) + debug(" dispatching #{event.message} to #{q}") if $options[:debug] + q.publish(event.message) + end + +end + +$options = { + :address => "localhost:5672", + :debug => false +} + +OptionParser.new do |opts| + opts.banner = "Usage: #{Pathname.new(__FILE__).basename} [$options]" + + opts.on("-a", "--address=ADDRESS", "Send messages to ADDRESS (def. #{$options[:address]}).") do |address| + $options[:address] = address + end + + opts.on("-d", "--debug", "Enable debugging output (def. #{$options[:debug]})") do + $options[:debug] = true + end + +end.parse! + +begin + Qpid::Proton::Reactor::Container.new(Broker.new($options[:address])).run +rescue Interrupt +end http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/reactor/client.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/reactor/client.rb b/examples/ruby/reactor/client.rb new file mode 100644 index 0000000..8bb58da --- /dev/null +++ b/examples/ruby/reactor/client.rb @@ -0,0 +1,68 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +require 'qpid_proton' + +class Client < Qpid::Proton::Handler::MessagingHandler + + def initialize(url, requests) + super() + @url = url + @requests = requests + end + + def on_start(event) + @sender = event.container.create_sender(@url) + @receiver = event.container.create_receiver(@sender.connection, :dynamic => true) + end + + def next_request + if @receiver.remote_source.address + req = Qpid::Proton::Message.new + req.reply_to = @receiver.remote_source.address + req.body = @requests.first + puts "-> #{req.body}" + @sender.send(req) + end + end + + def on_link_opened(event) + if event.receiver == @receiver + next_request + end + end + + def on_message(event) + puts "<- #{event.message.body}" + @requests.delete_at(0) + if !@requests.empty? + next_request + else + event.connection.close + end + end + +end + +REQUESTS = ["Twas brillig, and the slithy toves", + "Did gire and gymble in the wabe.", + "All mimsy were the borogroves,", + "And the mome raths outgrabe."] + +Qpid::Proton::Reactor::Container.new(Client.new("0.0.0.0:5672/examples", REQUESTS)).run http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/reactor/direct_recv.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/reactor/direct_recv.rb b/examples/ruby/reactor/direct_recv.rb new file mode 100644 index 0000000..2e19b04 --- /dev/null +++ b/examples/ruby/reactor/direct_recv.rb @@ -0,0 +1,60 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +require 'qpid_proton' +require 'optparse' + +require_relative '../lib/send_and_receive' + +class DirectReceive < ExampleReceive + + def initialize(url, expected) + super + end + + def on_start(event) + @acceptor = event.container.listen(self.url) + end + + def on_message(event) + super(event) + @acceptor.close if self.finished? + end + +end + +options = { + :address => "localhost:5672/examples", + :messages => 100, +} + +OptionParser.new do |opts| + opts.banner = "Usage: simple_send.rb [options]" + + opts.on("-a", "--address=ADDRESS", "Send messages to ADDRESS (def. #{options[:address]}).") do |address| + options[:address] = address + end + + opts.on("-m", "--messages=COUNT", "The number of messages to send (def. #{options[:messages]}", + OptionParser::DecimalInteger) do |messages| + options[:messages] = messages + end +end.parse! + +Qpid::Proton::Reactor::Container.new(DirectReceive.new(options[:address], options[:messages])).run http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/reactor/direct_send.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/reactor/direct_send.rb b/examples/ruby/reactor/direct_send.rb new file mode 100644 index 0000000..22ce7de --- /dev/null +++ b/examples/ruby/reactor/direct_send.rb @@ -0,0 +1,59 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +require 'qpid_proton' +require 'optparse' + +require_relative '../lib/send_and_receive' + +options = { + :address => "localhost:5672/examples", + :messages => 100, +} + +class SimpleSend < ExampleSend + + def initialize(url, messages) + super(url, messages) + end + + def on_start(event) + @acceptor = event.container.listen(url) + end + +end + +OptionParser.new do |opts| + opts.banner = "Usage: simple_send.rb [options]" + + opts.on("-a", "--address=ADDRESS", "Send messages to ADDRESS (def. #{options[:address]}).") do |address| + options[:address] = address + end + + opts.on("-m", "--messages=COUNT", "The number of messages to send (def. #{options[:messages]}", + OptionParser::DecimalInteger) do |messages| + options[:messages] = messages + end +end.parse! + +begin + Qpid::Proton::Reactor::Container.new(SimpleSend.new(options[:address], options[:messages])).run +rescue Interrupt => error + puts "ERROR: #{error}" +end http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/reactor/helloworld.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/reactor/helloworld.rb b/examples/ruby/reactor/helloworld.rb new file mode 100644 index 0000000..03eb561 --- /dev/null +++ b/examples/ruby/reactor/helloworld.rb @@ -0,0 +1,69 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +require 'qpid_proton' +require 'optparse' + +class HelloWorld < Qpid::Proton::Handler::MessagingHandler + + def initialize(server, address) + super() + @server = server + @address = address + end + + def on_start(event) + conn = event.container.connect(:address => @server) + event.container.create_sender(conn, :target => @address) + event.container.create_receiver(conn, :source => @address) + end + + def on_sendable(event) + msg = Qpid::Proton::Message.new + msg.body = "Hello world!" + event.sender.send(msg) + event.sender.close + end + + def on_message(event) + puts event.message.body + event.connection.close + end +end + +options = { + :address => "localhost:5672", + :queue => "examples" +} + +OptionParser.new do |opts| + opts.banner = "Usage: helloworld_direct.rb [options]" + + opts.on("-a", "--address=ADDRESS", "Send messages to ADDRESS (def. #{options[:address]}).") do |address| + options[:address] = address + end + + opts.on("-q", "--queue=QUEUE", "Send messages to QUEUE (def. #{options[:queue]})") do |queue| + options[:queue] = queue + end + +end.parse! + +hw = HelloWorld.new(options[:address], "examples") +Qpid::Proton::Reactor::Container.new(hw).run http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/reactor/helloworld_direct.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/reactor/helloworld_direct.rb b/examples/ruby/reactor/helloworld_direct.rb new file mode 100644 index 0000000..e98cc1f --- /dev/null +++ b/examples/ruby/reactor/helloworld_direct.rb @@ -0,0 +1,74 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +require 'qpid_proton' +require 'optparse' + +options = { + :address => "localhost:5672/examples", +} + +class HelloWorldDirect < Qpid::Proton::Handler::MessagingHandler + + include Qpid::Proton::Util::Wrapper + + def initialize(url) + super() + @url = url + end + + def on_start(event) + @acceptor = event.container.listen(@url) + event.container.create_sender(@url) + end + + def on_sendable(event) + msg = Qpid::Proton::Message.new + msg.body = "Hello world!" + event.sender.send(msg) + event.sender.close + end + + def on_message(event) + puts "#{event.message.body}" + end + + def on_accepted(event) + event.connection.close + end + + def on_connection_closed(event) + @acceptor.close + end + +end + +OptionParser.new do |opts| + opts.banner = "Usage: helloworld_direct.rb [options]" + + opts.on("-a", "--address=ADDRESS", "Send messages to ADDRESS (def. #{options[:address]}).") do |address| + options[:address] = address + end + +end.parse! + +begin + Qpid::Proton::Reactor::Container.new(HelloWorldDirect.new(options[:address])).run +rescue Interrupt => error +end http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/reactor/server.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/reactor/server.rb b/examples/ruby/reactor/server.rb new file mode 100644 index 0000000..e149dba --- /dev/null +++ b/examples/ruby/reactor/server.rb @@ -0,0 +1,64 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +require 'qpid_proton' + +class Server < Qpid::Proton::Handler::MessagingHandler + + def initialize(url, address) + super() + @url = url + @address = address + @senders = {} + end + + def on_start(event) + puts "Listening on #{@url}" + @container = event.container + @conn = @container.connect(:address => @url) + @receiver = @container.create_receiver(@conn, :source => @address) + @relay = nil + end + + def on_connection_opened(event) + if event.connection.remote_offered_capabilities && + event.connection.remote_offered_capabilities.contain?("ANONYMOUS-RELAY") + @relay = @container.create_sender(@conn, nil) + end + end + + def on_message(event) + msg = event.message + puts "<- #{msg.body}" + sender = @relay || @senders[msg.reply_to] + if sender.nil? + sender = @container.create_sender(@conn, :target => msg.reply_to) + @senders[msg.reply_to] = sender + end + reply = Qpid::Proton::Message.new + reply.address = msg.reply_to + reply.body = msg.body.upcase + puts "-> #{reply.body}" + reply.correlation_id = msg.correlation_id + sender.send(reply) + end + +end + +Qpid::Proton::Reactor::Container.new(Server.new("0.0.0.0:5672", "examples")).run() http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/reactor/simple_recv.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/reactor/simple_recv.rb b/examples/ruby/reactor/simple_recv.rb new file mode 100644 index 0000000..55a37ee --- /dev/null +++ b/examples/ruby/reactor/simple_recv.rb @@ -0,0 +1,58 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +require 'qpid_proton' +require 'optparse' + +require_relative '../lib/send_and_receive' + +class Receiver < ExampleReceive + + def initialize(url, count) + super(url, count) + end + + def on_start(event) + event.container.create_receiver(@url) + end + +end + +options = { + :address => "localhost:5672/examples", + :messages => 100, +} + +OptionParser.new do |opts| + opts.banner = "Usage: simple_send.rb [options]" + + opts.on("-a", "--address=ADDRESS", "Send messages to ADDRESS (def. #{options[:address]}).") do |address| + options[:address] = address + end + + opts.on("-m", "--messages=COUNT", "The number of messages to send (def. #{options[:messages]}", + OptionParser::DecimalInteger) do |messages| + options[:messages] = messages + end +end.parse! + +begin + Qpid::Proton::Reactor::Container.new(Receiver.new(options[:address], options[:messages])).run +rescue Interrupt +end http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/f63454c5/examples/ruby/reactor/simple_send.rb ---------------------------------------------------------------------- diff --git a/examples/ruby/reactor/simple_send.rb b/examples/ruby/reactor/simple_send.rb new file mode 100644 index 0000000..1dd4150 --- /dev/null +++ b/examples/ruby/reactor/simple_send.rb @@ -0,0 +1,55 @@ +#-- +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +#++ + +require 'qpid_proton' +require 'optparse' + +require_relative '../lib/send_and_receive' + +options = { + :address => "localhost:5672/examples", + :messages => 100, +} + +class SimpleSend < ExampleSend + + def initialize(url, messages) + super(url, messages) + end + + def on_start(event) + event.container.create_sender(url) + end + +end + +OptionParser.new do |opts| + opts.banner = "Usage: simple_send.rb [options]" + + opts.on("-a", "--address=ADDRESS", "Send messages to ADDRESS (def. #{options[:address]}).") do |address| + options[:address] = address + end + + opts.on("-m", "--messages=COUNT", "The number of messages to send (def. #{options[:messages]}", + OptionParser::DecimalInteger) do |messages| + options[:messages] = messages + end +end.parse! + +Qpid::Proton::Reactor::Container.new(SimpleSend.new(options[:address], options[:messages])).run --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org For additional commands, e-mail: commits-h...@qpid.apache.org
