An AMQP container manages a set of {Listener}s and {Connection}s which contain {#Sender} and {#Receiver} links to transfer messages. Usually, each AMQP client or server process has a single container for all of its connections and links.
One or more threads can call {#run}, events generated by all the listeners and connections will be dispatched in the {#run} threads.
Auto-stop flag.
True (the default) means that the container will stop automatically, as if {#stop} had been called, when the last listener or connection closes.
False means {#run} will not return unless {#stop} is called.
@return [Bool] auto-stop state
@return [MessagingHandler] The container-wide handler
@return [String] unique identifier for this container
True if the container has been stopped and can no longer be used. @return [Bool] stopped state
Create a new Container @overload initialize(id=nil)
@param id [String,Symbol] A unique ID for this container, use random UUID if nil.
@overload initialize(handler=nil, id=nil)
@param id [String,Symbol] A unique ID for this container, use random UUID if nil. @param handler [MessagingHandler] Optional default handler for connections that do not have their own handler (see {#connect} and {#listen}) *Note*: For multi-threaded code, it is recommended to use a separate handler instance for each connection, as a shared handler may be called concurrently.
# File lib/core/container.rb, line 55 def initialize(*args) @handler, @id, @panic = nil case args.size when 2 then @handler, @id = args when 1 then @id = String.try_convert(args[0]) || (args[0].to_s if args[0].is_a? Symbol) @handler = args[0] unless @id when 0 then else raise ArgumentError, "wrong number of arguments (given #{args.size}, expected 0..2" end # Use an empty messaging adapter to give default behaviour if there's no global handler. @adapter = Handler::Adapter.adapt(@handler) || Handler::MessagingAdapter.new(nil) @id = (@id || SecureRandom.uuid).freeze # Implementation note: # # - #run threads take work items from @work, process them, and rearm them for select # - work items are: ConnectionTask, ListenTask, :start, :select, :schedule # - nil on the @work queue makes a #run thread exit @work = Queue.new @work << :start @work << :select @wake = SelectWaker.new # Wakes #run thread in IO.select @auto_stop = true # Stop when @active drops to 0 @schedule = Schedule.new @schedule_working = false # True if :schedule is on the work queue # Following instance variables protected by lock @lock = Mutex.new @active = 0 # All active tasks, in @selectable, @work or being processed @selectable = Set.new # Tasks ready to block in IO.select @running = 0 # Count of #run threads @stopped = false # #stop called @stop_err = nil # Optional error to pass to tasks, from #stop end
Open an AMQP connection.
@param url [String, URI] Open a {TCPSocket} to url.host, url.port. url.scheme must be “amqp” or “amqps”, url.scheme.nil? is treated as “amqp” url.user, url.password are used as defaults if opts, opts are nil @option (see Qpid::Proton::Connection#open) @return [Connection] The new AMQP connection
# File lib/core/container.rb, line 123 def connect(url, opts=nil) not_stopped url = Qpid::Proton::uri url opts ||= {} if url.user || url.password opts[:user] ||= url.user opts[:password] ||= url.password end opts[:ssl_domain] ||= SSLDomain.new(SSLDomain::MODE_CLIENT) if url.scheme == "amqps" connect_io(TCPSocket.new(url.host, url.port), opts) end
Open an AMQP protocol connection on an existing {IO} object @param io [IO] An existing {IO} object, e.g. a {TCPSocket} @option (see Qpid::Proton::Connection#open)
# File lib/core/container.rb, line 138 def connect_io(io, opts=nil) not_stopped cd = connection_driver(io, opts) cd.connection.open() add(cd) cd.connection end
Listen for incoming AMQP connections
@param url [String,URI] Listen on host:port of the AMQP URL @param handler [Listener::Handler] A {Listener::Handler} object that will be called with events for this listener and can generate a new set of options for each one. @return [Listener] The AMQP listener.
# File lib/core/container.rb, line 153 def listen(url, handler=Listener::Handler.new) not_stopped url = Qpid::Proton::uri url # TODO aconway 2017-11-01: amqps, SSL listen_io(TCPServer.new(url.host, url.port), handler) end
Listen for incoming AMQP connections on an existing server socket. @param io A server socket, for example a {TCPServer} @param handler [Listener::Handler] Handler for events from this listener
# File lib/core/container.rb, line 164 def listen_io(io, handler=Listener::Handler.new) not_stopped l = ListenTask.new(io, handler, self) add(l) l end
Run the container: wait for IO activity, dispatch events to handlers.
Multi-threaading : More than one thread can call {#run} concurrently, the container will use all {#run} threads as a thread pool. Calls to {MessagingHandler} or {Listener::Handler} methods are serialized for each connection or listener. See {WorkQueue} for coordinating with other threads.
Exceptions: If any handler method raises an exception it will stop the container, and the exception will be raised by all calls to {#run}. For single threaded code this is often desirable. Multi-threaded server applications should normally rescue exceptions in the handler and deal with them in another way: logging, closing the connection with an error condition, signalling another thread etc.
@return [void] Returns when the container stops, see {#stop} and {#auto_stop}
@raise [StoppedError] If the container has already been stopped when {#run} was called.
@raise [Exception] If any {MessagingHandler} or {Listener::Handler} managed by
the container raises an exception, that exception will be raised by {#run}
# File lib/core/container.rb, line 193 def run @lock.synchronize do @running += 1 # Note: ensure clause below will decrement @running raise StoppedError if @stopped end while task = @work.pop run_one(task, Time.now) end raise @panic if @panic ensure @lock.synchronize do if (@running -= 1) > 0 work_wake nil # Signal the next thread else @adapter.on_container_stop(self) if @adapter.respond_to? :on_container_stop end end end
Number of threads in {#run} @return [Bool] {#run} thread count
# File lib/core/container.rb, line 114 def running() @lock.synchronize { @running }; end
Schedule code to be executed after a delay.
@param delay [Numeric] delay in seconds, must be >= 0 @yield [ ] the
block is invoked with no parameters in a {#run} thread after
delay
has elapsed @return [void] @raise [ThreadError] if
non_block
is true and the operation would block
# File lib/core/container.rb, line 250 def schedule(delay, non_block=false, &block) not_stopped @lock.synchronize { @active += 1 } if @schedule.add(Time.now + delay, non_block, &block) @wake.wake end
Stop the container.
Close all listeners and abort all connections without doing AMQP protocol close.
{#stop} returns immediately, calls to {#run} will return when all activity is finished.
The container can no longer be used, using a stopped container raises {StoppedError} on attempting. Create a new container if you want to resume activity.
@param error [Condition] Optional error condition passed to
{MessagingHandler#on_transport_error} for each connection and {Listener::Handler::on_error} for each listener.
@param panic [Exception] Optional exception raised by all concurrent calls to run()
# File lib/core/container.rb, line 230 def stop(error=nil, panic=nil) @lock.synchronize do return if @stopped @stop_err = Condition.convert(error) @panic = panic @stopped = true check_stop_lh # NOTE: @stopped => # - no new run threads can join # - no more select calls after next wakeup # - once @active == 0, all threads will be stopped with nil end @wake.wake end
# File lib/core/container.rb, line 258 def wake() @wake.wake; end