| # 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 'thread' |
| require 'set' |
| require_relative 'listener' |
| require_relative 'work_queue' |
| |
| module Qpid::Proton |
| public |
| |
| # 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. |
| class Container |
| include TimeCompare |
| |
| # Error raised if the container is used after {#stop} has been called. |
| class StoppedError < Qpid::Proton::StoppedError |
| def initialize() super("container has been stopped"); end |
| end |
| |
| # 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. |
| # |
| def initialize(*args) |
| @handler, @id = 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 |
| |
| # Threading and implementation notes: see comment on #run_one |
| @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 |
| @work_queue = WorkQueue.new(self) # work scheduled by other threads for :select context |
| |
| # 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 |
| @panic = nil # Exception caught in a run thread, to be raised by all run threads |
| end |
| |
| # @return [MessagingHandler] The container-wide handler |
| attr_reader :handler |
| |
| # @return [String] unique identifier for this container |
| attr_reader :id |
| |
| def to_s() "#<#{self.class} id=#{id.inspect}>"; end |
| def inspect() to_s; end |
| |
| # 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 |
| attr_accessor :auto_stop |
| |
| # True if the container has been stopped and can no longer be used. |
| # @return [Bool] stopped state |
| attr_accessor :stopped |
| |
| # Number of threads in {#run} |
| # @return [Bool] {#run} thread count |
| def running() @lock.synchronize { @running }; 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[:user], opts[:password] are nil |
| # @option (see Connection#open) |
| # @return [Connection] The new AMQP connection |
| 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 Connection#open) |
| 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. |
| # |
| 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 |
| # |
| def listen_io(io, handler=Listener::Handler.new) |
| not_stopped |
| l = ListenTask.new(io, handler, self) |
| add(l) |
| l.listener |
| 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} |
| # |
| 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 |
| @lock.synchronize { raise @panic if @panic } |
| ensure |
| @lock.synchronize do |
| if (@running -= 1) > 0 |
| work_wake nil # Signal the next thread |
| else |
| # This is the last thread, no need to do maybe_panic around this final handler call. |
| @adapter.on_container_stop(self) if @adapter.respond_to? :on_container_stop |
| end |
| end |
| 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}. 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 to raise from all calls to run() |
| # |
| 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 |
| end |
| |
| # Get the {WorkQueue} that can be used to schedule code to be run by the container. |
| # |
| # Note: to run code that affects a {Connection} or it's associated objects, |
| # use {Connection#work_queue} |
| def work_queue() @work_queue; end |
| |
| # (see WorkQueue#schedule) |
| def schedule(at, &block) @work_queue.schedule(at, &block) end |
| |
| private |
| |
| def wake() @wake.wake; end |
| |
| class ConnectionTask < Qpid::Proton::HandlerDriver |
| include TimeCompare |
| |
| def initialize container, io, opts, server=false |
| super io, opts[:handler] |
| transport.set_server if server |
| transport.apply opts |
| connection.apply opts |
| @work_queue = WorkQueue.new(container) |
| connection.instance_variable_set(:@work_queue, @work_queue) |
| end |
| def next_tick() earliest(super, @work_queue.next_tick); end |
| def process(now) @work_queue.process(now); super(); end |
| |
| def dispatch # Intercept dispatch to close work_queue |
| super |
| @work_queue.close if read_closed? && write_closed? |
| end |
| end |
| |
| class ListenTask < Listener |
| |
| def initialize(io, handler, container) |
| @io, @handler = io, handler |
| @listener = Listener.new(io, container) |
| env = ENV['PN_TRACE_EVT'] |
| if env && ["true", "1", "yes", "on"].include?(env.downcase) |
| @log_prefix = "[0x#{object_id.to_s(16)}](PN_LISTENER_" |
| else |
| @log_prefix = nil |
| end |
| dispatch(:on_open); |
| end |
| |
| attr_reader :listener |
| def closing?() @listener.instance_variable_get(:@closing); end |
| def condition() @listener.instance_variable_get(:@condition); end |
| def closed?() @io.closed?; end |
| |
| def process |
| return if closed? |
| unless closing? |
| begin |
| return @io.accept, dispatch(:on_accept) |
| rescue IO::WaitReadable, Errno::EINTR |
| rescue StandardError => e |
| @listener.close(e) |
| end |
| end |
| ensure |
| if closing? |
| @io.close rescue nil |
| @listener.instance_variable_set(:@closed, true) |
| dispatch(:on_error, condition) if condition |
| dispatch(:on_close) |
| end |
| end |
| |
| def can_read?() !finished?; end |
| def can_write?() false; end |
| def finished?() closed?; end |
| |
| def dispatch(method, *args) |
| # TODO aconway 2017-11-27: better logging |
| STDERR.puts "#{@log_prefix}#{([method[3..-1].upcase]+args).join ', '})" if @log_prefix |
| @handler.__send__(method, self, *args) if @handler && @handler.respond_to?(method) |
| end |
| |
| def next_tick() nil; end |
| |
| # Close listener and force immediate close of socket |
| def close(e=nil) |
| @listener.close(e) |
| @io.close rescue nil |
| end |
| end |
| |
| # Selectable object that can be used to wake IO.select from another thread |
| class SelectWaker |
| def initialize |
| @rd, @wr = IO.pipe |
| @lock = Mutex.new |
| @set = false |
| end |
| |
| def to_io() @rd; end |
| |
| def wake |
| @lock.synchronize do |
| return if @set # Don't write if already has data |
| @set = true |
| @wr.write_nonblock('x') rescue nil |
| end |
| end |
| |
| def reset |
| @lock.synchronize do |
| return unless @set |
| @rd.read_nonblock(1) rescue nil |
| @set = false |
| end |
| end |
| |
| def close |
| @rd.close |
| @wr.close |
| end |
| end |
| |
| # Handle a single item from the @work queue, this is the heart of the #run loop. |
| # Take one task from @work, process it, and rearm for select |
| # Tasks are: ConnectionTask, ListenTask, :start, :select |
| # - ConnectionTask/ListenTask have #can_read, #can_write, #next_tick to set up IO.select |
| # and #process to run handlers and process relevant work_queue |
| # - nil means exit from the #run thread exit (handled by #run) |
| # - :select does IO.select and processes Container#work_queue |
| def run_one(task, now) |
| case task |
| |
| when :start |
| maybe_panic { @adapter.on_container_start(self) } if @adapter.respond_to? :on_container_start |
| |
| when :select |
| # Compute read/write select sets and minimum next_tick for select timeout |
| r, w = [@wake], [] |
| next_tick = @work_queue.next_tick |
| @lock.synchronize do |
| @selectable.each do |s| |
| r << s if s.can_read? |
| w << s if s.can_write? |
| next_tick = earliest(s.next_tick, next_tick) |
| end |
| end |
| |
| timeout = ((next_tick > now) ? next_tick - now : 0) if next_tick |
| r, w = IO.select(r, w, nil, timeout) |
| @wake.reset if r && r.delete(@wake) |
| now = Time.now unless timeout == 0 # Update now if we may have blocked |
| |
| # selected is a Set to eliminate duplicates between r, w and next_tick due. |
| selected = Set.new |
| selected.merge(r) if r |
| selected.merge(w) if w |
| stopped = @lock.synchronize do |
| if @stopped # close everything |
| @selectable.each { |s| s.close @stop_err; @work << s } |
| @selectable.clear |
| @work_queue.close |
| @wake.close |
| else |
| @selectable -= selected # Remove already-selected tasks from @selectable |
| # Also select and remove items with next_tick before now |
| @selectable.delete_if { |s| before_eq(s.next_tick, now) and selected << s } |
| end |
| @stopped |
| end |
| selected.each { |s| @work << s } # Queue up tasks needing #process |
| maybe_panic { @work_queue.process(now) } # Process current work queue items |
| @work_queue.clear if stopped |
| @lock.synchronize { check_stop_lh } if @work_queue.empty? |
| |
| @work << :select unless stopped # Enable next select |
| |
| when ConnectionTask then |
| maybe_panic { task.process now } |
| rearm task |
| |
| when ListenTask then |
| io, opts = maybe_panic { task.process } |
| add(connection_driver(io, opts, true)) if io |
| rearm task |
| end |
| end |
| |
| # Rescue any exception raised by the block and stop the container. |
| def maybe_panic |
| begin |
| yield |
| rescue Exception => e |
| stop(nil, e) |
| nil |
| end |
| end |
| |
| # Normally if we add work we need to set a wakeup to ensure a single #run |
| # thread doesn't get stuck in select while there is other work on the queue. |
| def work_wake(task) |
| @work << task |
| @wake.wake |
| end |
| |
| def connection_driver(io, opts=nil, server=false) |
| opts ||= {} |
| opts[:container] = self |
| opts[:handler] ||= @adapter |
| ConnectionTask.new(self, io, opts, server) |
| end |
| |
| # All new tasks are added here |
| def add task |
| @lock.synchronize do |
| @active += 1 |
| task.close @stop_err if @stopped |
| end |
| work_wake task |
| end |
| |
| def rearm task |
| @lock.synchronize do |
| if task.finished? |
| @active -= 1 |
| check_stop_lh |
| elsif @stopped |
| task.close @stop_err |
| work_wake task |
| else |
| @selectable << task |
| end |
| end |
| @wake.wake |
| end |
| |
| def check_stop_lh |
| if @active.zero? && (@auto_stop || @stopped) && @work_queue.empty? |
| @stopped = true |
| work_wake nil # Signal threads to stop |
| true |
| end |
| end |
| |
| def not_stopped() raise StoppedError if @lock.synchronize { @stopped }; end |
| |
| end |
| end |
