| # 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. |
| # |
| |
| # Console API for Qpid Management Framework |
| |
| require 'socket' |
| require 'monitor' |
| require 'thread' |
| require 'uri' |
| require 'time' |
| |
| module Qpid::Qmf |
| |
| # To access the asynchronous operations, a class must be derived from |
| # Console with overrides of any combination of the available methods. |
| class Console |
| |
| # Invoked when a connection is established to a broker |
| def broker_connected(broker); end |
| |
| # Invoked when the connection to a broker is lost |
| def broker_disconnected(broker); end |
| |
| # Invoked when a QMF package is discovered |
| def new_package(name); end |
| |
| # Invoked when a new class is discovered. Session.getSchema can be |
| # used to obtain details about the class |
| def new_class(kind, klass_key); end |
| |
| # Invoked when a QMF agent is discovered |
| def new_agent(agent); end |
| |
| # Invoked when a QMF agent disconects |
| def del_agent(agent); end |
| |
| # Invoked when an object is updated |
| def object_props(broker, record); end |
| |
| # Invoked when an object is updated |
| def object_stats(broker, record); end |
| |
| # Invoked when an event is raised |
| def event(broker, event); end |
| |
| # Invoked when an agent heartbeat is received. |
| def heartbeat(agent, timestamp); end |
| |
| # Invoked when the connection sequence reaches the point where broker information is available. |
| def broker_info(broker); end |
| |
| # Invoked when a method response from an asynchronous method call is received. |
| def method_response(broker, seq, response); end |
| end |
| |
| class BrokerURL |
| |
| attr_reader :host, :port, :auth_name, :auth_pass |
| |
| def initialize(text) |
| uri = URI.parse(text) |
| |
| @host = uri.host |
| @port = uri.port ? uri.port : 5672 |
| @auth_name = uri.user |
| @auth_pass = uri.password |
| |
| return uri |
| end |
| |
| def name |
| "#{@host}:#{@port}" |
| end |
| |
| def match(host, port) |
| # FIXME: Unlcear what the Python code is actually checking for |
| # here, especially since HOST can resolve to multiple IP's |
| @port == port && |
| (host == @host || ipaddr(host, port) == ipaddr(@host, @port)) |
| end |
| |
| private |
| def ipaddr(host, port) |
| s = Socket::getaddrinfo(host, port, |
| Socket::AF_INET, Socket::SOCK_STREAM) |
| s[0][2] |
| end |
| end |
| |
| # An instance of the Session class represents a console session running |
| # against one or more QMF brokers. A single instance of Session is |
| # needed to interact with the management framework as a console. |
| class Session |
| CONTEXT_SYNC = 1 |
| CONTEXT_STARTUP = 2 |
| CONTEXT_MULTIGET = 3 |
| |
| DEFAULT_GET_WAIT_TIME = 60 |
| |
| include MonitorMixin |
| |
| attr_reader :binding_key_list, :select, :seq_mgr, :console, :packages |
| |
| # Initialize a session. If the console argument is provided, the |
| # more advanced asynchronous features are available. If console is |
| # defaulted, the session will operate in a simpler, synchronous |
| # manner. The rcvObjects, rcvEvents, and rcvHeartbeats arguments |
| # are meaningful only if 'console' is provided. They control |
| # whether object updates, events, and agent-heartbeats are |
| # subscribed to. If the console is not interested in receiving one |
| # or more of the above, setting the argument to False will reduce |
| # tha bandwidth used by the API. If manageConnections is set to |
| # True, the Session object will manage connections to the brokers. |
| # This means that if a broker is unreachable, it will retry until a |
| # connection can be established. If a connection is lost, the |
| # Session will attempt to reconnect. |
| # |
| # If manageConnections is set to False, the user is responsible for |
| # handing failures. In this case, an unreachable broker will cause |
| # addBroker to raise an exception. If userBindings is set to False |
| # (the default) and rcvObjects is True, the console will receive |
| # data for all object classes. If userBindings is set to True, the |
| # user must select which classes the console shall receive by |
| # invoking the bindPackage or bindClass methods. This allows the |
| # console to be configured to receive only information that is |
| # relavant to a particular application. If rcvObjects id False, |
| # userBindings has no meaning. |
| # |
| # Accept a hash of parameters, where keys can be :console, |
| # :rcv_objects, :rcv_events, :rcv_heartbeats, :manage_connections, |
| # and :user_bindings |
| def initialize(kwargs = {}) |
| super() |
| @console = kwargs[:console] || nil |
| @brokers = [] |
| @packages = {} |
| @seq_mgr = SequenceManager.new |
| @cv = new_cond |
| @sync_sequence_list = [] |
| @result = [] |
| @select = [] |
| @error = nil |
| @rcv_objects = kwargs[:rcv_objects] == nil ? true : kwargs[:rcv_objects] |
| @rcv_events = kwargs[:rcv_events] == nil ? true : kwargs[:rcv_events] |
| @rcv_heartbeats = kwargs[:rcv_heartbeats] == nil ? true : kwargs[:rcv_heartbeats] |
| @user_bindings = kwargs[:user_bindings] == nil ? false : kwargs[:user_bindings] |
| unless @console |
| @rcv_objects = false |
| @rcv_events = false |
| @rcv_heartbeats = false |
| end |
| @binding_key_list = binding_keys |
| @manage_connections = kwargs[:manage_connections] || false |
| |
| if @user_bindings && ! @rcv_objects |
| raise ArgumentError, "user_bindings can't be set unless rcv_objects is set and a console is provided" |
| end |
| |
| end |
| |
| def to_s |
| "QMF Console Session Manager (brokers: #{@brokers.size})" |
| end |
| |
| def managedConnections? |
| return @manage_connections |
| end |
| |
| # Connect to a Qpid broker. Returns an object of type Broker |
| # |
| # To supply a username for authentication, use the URL syntax: |
| # |
| # amqp://username@hostname:port |
| # |
| # If the broker needs a password for the client, an interactive prompt will be |
| # provided to the user. |
| # |
| # To supply a username and a password, use |
| # |
| # amqp://username:password@hostname:port |
| # |
| # The following keyword arguments may be used to control authentication: |
| # |
| # :mechanism - SASL mechanism (i.e. "PLAIN", "GSSAPI", "ANONYMOUS", etc. |
| # - defaults to unspecified (the system chooses for you) |
| # :service - SASL service name (i.e. the kerberos principal of the broker) |
| # - defaults to "qpidd" |
| # :min_ssf - Minimum Security Strength Factor for SASL security layers |
| # - defaults to 0 |
| # :max_ssf - Maximum Security Strength Factor for SASL security layers |
| # - defaults to 65535 |
| # |
| def add_broker(target = "amqp://localhost", kwargs = {}) |
| url = BrokerURL.new(target) |
| broker = Broker.new(self, url.host, url.port, url.auth_name, url.auth_pass, kwargs) |
| unless broker.connected? || @manage_connections |
| raise broker.error |
| end |
| |
| @brokers << broker |
| objects(:broker => broker, :class => "agent") unless @manage_connections |
| return broker |
| end |
| |
| # Disconnect from a broker. The 'broker' argument is the object |
| # returned from the addBroker call |
| def del_broker(broker) |
| broker.shutdown |
| @brokers.delete(broker) |
| end |
| |
| # Get the list of known classes within a QMF package |
| def classes(package_name) |
| list = [] |
| @brokers.each { |broker| broker.wait_for_stable } |
| if @packages.include?(package_name) |
| # FIXME What's the actual structure of @packages[package_name] |
| @packages[package_name].each do |key, schema_class| |
| list << schema_class.klass_key |
| end |
| end |
| return list |
| end |
| |
| # Get the schema for a QMF class |
| def schema(klass_key) |
| @brokers.each { |broker| broker.wait_for_stable } |
| if @packages.include?(klass_key.package) |
| @packages[klass_key.package][ [klass_key.klass_name, klass_key.hash] ] |
| end |
| end |
| |
| def bind_package(package_name) |
| unless @user_bindings && @rcv_objects |
| raise "userBindings option not set for Session" |
| end |
| @brokers.each do |broker| |
| args = { :exchange => "qpid.management", |
| :queue => broker.topic_name, |
| :binding_key => "console.obj.*.*.#{package_name}.#" } |
| broker.amqp_session.exchange_bind(args) |
| end |
| end |
| |
| def bind_class(package_name, class_name) |
| unless @user_bindings && @rcv_objects |
| raise "userBindings option not set for Session" |
| end |
| @brokers.each do |broker| |
| args = { :exchange => "qpid.management", |
| :queue => broker.topic_name, |
| :binding_key=> "console.obj.*.*.#{package_name}.#{class_name}.#" } |
| broker.amqp_session.exchange_bind(args) |
| end |
| end |
| |
| def bind_class_key(klass_key) |
| unless @user_bindings && @rcv_objects |
| raise "userBindings option not set for Session" |
| end |
| pname, cname, hash = klass_key.to_a() |
| @brokers.each do |broker| |
| args = { :exchange => "qpid.management", |
| :queue => broker.topic_name, |
| :binding_key => "console.obj.*.*.#{pname}.#{cname}.#" } |
| broker.amqp_session.exchange_bind(args) |
| end |
| end |
| |
| # Get a list of currently known agents |
| def agents(broker=nil) |
| broker_list = [] |
| if broker.nil? |
| broker_list = @brokers.dup |
| else |
| broker_list << broker |
| end |
| broker_list.each { |b| b.wait_for_stable } |
| agent_list = [] |
| broker_list.each { |b| agent_list += b.agents } |
| return agent_list |
| end |
| |
| # Get a list of objects from QMF agents. |
| # All arguments are passed by name(keyword). |
| # |
| # The class for queried objects may be specified in one of the |
| # following ways: |
| # :schema => <schema> - supply a schema object returned from getSchema. |
| # :key => <key> - supply a klass_key from the list returned by getClasses. |
| # :class => <name> - supply a class name as a string. If the class name exists |
| # in multiple packages, a _package argument may also be supplied. |
| # :object_id = <id> - get the object referenced by the object-id |
| # |
| # If objects should be obtained from only one agent, use the following argument. |
| # Otherwise, the query will go to all agents. |
| # |
| # :agent = <agent> - supply an agent from the list returned by getAgents. |
| # |
| # If the get query is to be restricted to one broker (as opposed to |
| # all connected brokers), add the following argument: |
| # |
| # :broker = <broker> - supply a broker as returned by addBroker. |
| # |
| # The default timeout for this synchronous operation is 60 seconds. To change the timeout, |
| # use the following argument: |
| # |
| # :timeout = <time in seconds> |
| # |
| # If additional arguments are supplied, they are used as property |
| # selectors, as long as their keys are strings. For example, if |
| # the argument "name" => "test" is supplied, only objects whose |
| # "name" property is "test" will be returned in the result. |
| def objects(kwargs) |
| if kwargs.include?(:broker) |
| broker_list = [] |
| broker_list << kwargs[:broker] |
| else |
| broker_list = @brokers |
| end |
| broker_list.each { |broker| |
| broker.wait_for_stable |
| if kwargs[:package] != "org.apache.qpid.broker" or kwargs[:class] != "agent" |
| objects(:agent => broker.agent(1,0), :package => "org.apache.qpid.broker", :class => "agent") if broker.connected? |
| end |
| } |
| |
| agent_list = [] |
| if kwargs.include?(:agent) |
| agent = kwargs[:agent] |
| unless broker_list.include?(agent.broker) |
| raise ArgumentError, "Supplied agent is not accessible through the supplied broker" |
| end |
| agent_list << agent if agent.broker.connected? |
| else |
| if kwargs.include?(:object_id) |
| oid = kwargs[:object_id] |
| broker_list.each { |broker| |
| broker.agents.each { |agent| |
| if oid.broker_bank == agent.broker_bank && oid.agent_bank == agent.agent_bank |
| agent_list << agent if agent.broker.connected? |
| end |
| } |
| } |
| else |
| broker_list.each { |broker| |
| agent_list += broker.agents if broker.connected? |
| } |
| end |
| end |
| |
| cname = nil |
| if kwargs.include?(:schema) |
| # FIXME: What kind of object is kwargs[:schema] |
| pname, cname, hash = kwargs[:schema].getKey().to_a |
| elsif kwargs.include?(:key) |
| pname, cname, hash = kwargs[:key].to_a |
| elsif kwargs.include?(:class) |
| pname, cname, hash = [kwargs[:package], kwargs[:class], nil] |
| end |
| if cname.nil? && ! kwargs.include?(:object_id) |
| raise ArgumentError, |
| "No class supplied, use :schema, :key, :class, or :object_id' argument" |
| end |
| |
| map = {} |
| @select = [] |
| if kwargs.include?(:object_id) |
| map["_objectid"] = kwargs[:object_id].to_s |
| else |
| map["_class"] = cname |
| map["_package"] = pname if pname |
| map["_hash"] = hash if hash |
| kwargs.each do |k,v| |
| @select << [k, v] if k.is_a?(String) |
| end |
| end |
| |
| @result = [] |
| agent_list.each do |agent| |
| broker = agent.broker |
| send_codec = Qpid::StringCodec.new(broker.conn.spec) |
| seq = nil |
| synchronize do |
| seq = @seq_mgr.reserve(CONTEXT_MULTIGET) |
| @sync_sequence_list << seq |
| end |
| broker.set_header(send_codec, ?G, seq) |
| send_codec.write_map(map) |
| bank_key = "%d.%d" % [broker.broker_bank, agent.agent_bank] |
| smsg = broker.message(send_codec.encoded, "agent.#{bank_key}") |
| broker.emit(smsg) |
| end |
| |
| timeout = false |
| if kwargs.include?(:timeout) |
| wait_time = kwargs[:timeout] |
| else |
| wait_time = DEFAULT_GET_WAIT_TIME |
| end |
| synchronize do |
| unless @cv.wait_for(wait_time) { @sync_sequence_list.empty? || @error } |
| @sync_sequence_list.each do |pending_seq| |
| @seq_mgr.release(pending_seq) |
| end |
| @sync_sequence_list = [] |
| timeout = true |
| end |
| end |
| |
| if @error |
| errorText = @error |
| @error = nil |
| raise errorText |
| end |
| |
| if @result.empty? && timeout |
| raise "No agent responded within timeout period" |
| end |
| @result |
| end |
| |
| # Return one and only one object or nil. |
| def object(kwargs) |
| objs = objects(kwargs) |
| return objs.length == 1 ? objs[0] : nil |
| end |
| |
| # Return the first of potentially many objects. |
| def first_object(kwargs) |
| objs = objects(kwargs) |
| return objs.length > 0 ? objs[0] : nil |
| end |
| |
| def set_event_filter(kwargs); end |
| |
| def handle_broker_connect(broker); end |
| |
| def handle_broker_resp(broker, codec, seq) |
| broker.broker_id = codec.read_uuid |
| @console.broker_info(broker) if @console |
| |
| # Send a package request |
| # (effectively inc and dec outstanding by not doing anything) |
| send_codec = Qpid::StringCodec.new(broker.conn.spec) |
| seq = @seq_mgr.reserve(CONTEXT_STARTUP) |
| broker.set_header(send_codec, ?P, seq) |
| smsg = broker.message(send_codec.encoded) |
| broker.emit(smsg) |
| end |
| |
| def handle_package_ind(broker, codec, seq) |
| pname = codec.read_str8 |
| new_package = false |
| synchronize do |
| new_package = ! @packages.include?(pname) |
| @packages[pname] = {} if new_package |
| end |
| @console.new_package(pname) if @console |
| |
| # Send a class request |
| broker.inc_outstanding |
| send_codec = Qpid::StringCodec.new(broker.conn.spec) |
| seq = @seq_mgr.reserve(CONTEXT_STARTUP) |
| broker.set_header(send_codec, ?Q, seq) |
| send_codec.write_str8(pname) |
| smsg = broker.message(send_codec.encoded) |
| broker.emit(smsg) |
| end |
| |
| def handle_command_complete(broker, codec, seq) |
| code = codec.read_uint32 |
| text = codec.read_str8 |
| context = @seq_mgr.release(seq) |
| if context == CONTEXT_STARTUP |
| broker.dec_outstanding |
| elsif context == CONTEXT_SYNC && seq == broker.sync_sequence |
| broker.sync_done |
| elsif context == CONTEXT_MULTIGET && @sync_sequence_list.include?(seq) |
| synchronize do |
| @sync_sequence_list.delete(seq) |
| @cv.signal if @sync_sequence_list.empty? |
| end |
| end |
| end |
| |
| def handle_class_ind(broker, codec, seq) |
| kind = codec.read_uint8 |
| classKey = ClassKey.new(codec) |
| unknown = false |
| |
| synchronize do |
| return unless @packages.include?(classKey.package) |
| unknown = true unless @packages[classKey.package].include?([classKey.klass_name, classKey.hash]) |
| end |
| |
| |
| if unknown |
| # Send a schema request for the unknown class |
| broker.inc_outstanding |
| send_codec = Qpid::StringCodec.new(broker.conn.spec) |
| seq = @seq_mgr.reserve(CONTEXT_STARTUP) |
| broker.set_header(send_codec, ?S, seq) |
| classKey.encode(send_codec) |
| smsg = broker.message(send_codec.encoded) |
| broker.emit(smsg) |
| end |
| end |
| |
| def handle_method_resp(broker, codec, seq) |
| code = codec.read_uint32 |
| text = codec.read_str16 |
| out_args = {} |
| pair = @seq_mgr.release(seq) |
| return unless pair |
| method, synchronous = pair |
| if code == 0 |
| method.arguments.each do |arg| |
| if arg.dir.index(?O) |
| out_args[arg.name] = decode_value(codec, arg.type) |
| end |
| end |
| end |
| result = MethodResult.new(code, text, out_args) |
| if synchronous: |
| broker.synchronize do |
| broker.sync_result = MethodResult.new(code, text, out_args) |
| broker.sync_done |
| end |
| else |
| @console.method_response(broker, seq, result) if @console |
| end |
| end |
| |
| def handle_heartbeat_ind(broker, codec, seq, msg) |
| if @console |
| broker_bank = 1 |
| agent_bank = 0 |
| dp = msg.get("delivery_properties") |
| if dp |
| key = dp["routing_key"] |
| key_elements = key.split(".") |
| if key_elements.length == 4 |
| broker_bank = key_elements[2].to_i |
| agent_bank = key_elements[3].to_i |
| end |
| end |
| agent = broker.agent(broker_bank, agent_bank) |
| timestamp = codec.read_uint64 |
| @console.heartbeat(agent, timestamp) if agent |
| end |
| end |
| |
| def handle_event_ind(broker, codec, seq) |
| if @console |
| event = Event.new(self, broker, codec) |
| @console.event(broker, event) |
| end |
| end |
| |
| def handle_schema_resp(broker, codec, seq) |
| kind = codec.read_uint8 |
| classKey = ClassKey.new(codec) |
| klass = SchemaClass.new(self, kind, classKey, codec) |
| synchronize { @packages[classKey.package][ [classKey.klass_name, classKey.hash] ] = klass } |
| |
| @seq_mgr.release(seq) |
| broker.dec_outstanding |
| @console.new_class(kind, classKey) if @console |
| end |
| |
| def handle_content_ind(broker, codec, seq, prop=false, stat=false) |
| klass_key = ClassKey.new(codec) |
| pname, cname, hash = klass_key.to_a() ; |
| |
| schema = nil |
| synchronize do |
| return unless @packages.include?(klass_key.package) |
| return unless @packages[klass_key.package].include?([klass_key.klass_name, klass_key.hash]) |
| schema = @packages[klass_key.package][ [klass_key.klass_name, klass_key.hash] ] |
| end |
| |
| |
| object = Qpid::Qmf::Object.new(self, broker, schema, codec, prop, stat) |
| if pname == "org.apache.qpid.broker" && cname == "agent" && prop |
| broker.update_agent(object) |
| end |
| |
| synchronize do |
| if @sync_sequence_list.include?(seq) |
| if object.timestamps()[2] == 0 && select_match(object) |
| @result << object |
| end |
| return |
| end |
| end |
| |
| @console.object_props(broker, object) if @console && @rcv_objects && prop |
| @console.object_stats(broker, object) if @console && @rcv_objects && stat |
| end |
| |
| def handle_broker_disconnect(broker); end |
| |
| def handle_error(error) |
| synchronize do |
| @error = error if @sync_sequence_list.length > 0 |
| @sync_sequence_list = [] |
| @cv.signal |
| end |
| end |
| |
| # Decode, from the codec, a value based on its typecode |
| def decode_value(codec, typecode) |
| case typecode |
| when 1: data = codec.read_uint8 # U8 |
| when 2: data = codec.read_uint16 # U16 |
| when 3: data = codec.read_uint32 # U32 |
| when 4: data = codec.read_uint64 # U64 |
| when 6: data = codec.read_str8 # SSTR |
| when 7: data = codec.read_str16 # LSTR |
| when 8: data = codec.read_int64 # ABSTIME |
| when 9: data = codec.read_uint64 # DELTATIME |
| when 10: data = ObjectId.new(codec) # REF |
| when 11: data = codec.read_uint8 != 0 # BOOL |
| when 12: data = codec.read_float # FLOAT |
| when 13: data = codec.read_double # DOUBLE |
| when 14: data = codec.read_uuid # UUID |
| when 15: data = codec.read_map # FTABLE |
| when 16: data = codec.read_int8 # S8 |
| when 17: data = codec.read_int16 # S16 |
| when 18: data = codec.read_int32 # S32 |
| when 19: data = codec.read_int64 # S64 |
| when 20: # Object |
| inner_type_code = codec.read_uint8() |
| if (inner_type_code == 20) |
| classKey = ClassKey.new(codec) |
| innerSchema = schema(classKey) |
| data = Object.new(self, @broker, innerSchema, codec, true, true, false) if innerSchema |
| else |
| data = decode_value(codec, inner_type_code) |
| end |
| when 21: |
| data = [] |
| rec_codec = Qpid::StringCodec.new(codec.spec, codec.read_vbin32()) |
| count = rec_codec.read_uint32() |
| while count > 0 do |
| type = rec_codec.read_uint8() |
| data << (decode_value(rec_codec,type)) |
| count -= 1 |
| end |
| when 22: |
| data = [] |
| rec_codec = Qpid::StringCodec.new(codec.spec, codec.read_vbin32()) |
| count = rec_codec.read_uint32() |
| type = rec_codec.read_uint8() |
| while count > 0 do |
| data << (decode_value(rec_codec,type)) |
| count -= 1 |
| end |
| else |
| raise ArgumentError, "Invalid type code: #{typecode} - #{typecode.inspect}" |
| end |
| return data |
| end |
| |
| ENCODINGS = { |
| String => 6, |
| Fixnum => 18, |
| Bignum => 19, |
| Float => 12, |
| Array => 21, |
| Hash => 15 |
| } |
| |
| def encoding(object) |
| klass = object.class |
| if ENCODINGS.has_key?(klass) |
| return ENCODINGS[klass] |
| end |
| for base in klass.__bases__ |
| result = encoding(base) |
| return result unless result.nil? |
| end |
| end |
| |
| # Encode, into the codec, a value based on its typecode |
| def encode_value(codec, value, typecode) |
| # FIXME: Python does a lot of magic type conversions |
| # We just assume that value has the right type; this is safer |
| # than coercing explicitly, since Array::pack will complain |
| # loudly about various type errors |
| case typecode |
| when 1: codec.write_uint8(value) # U8 |
| when 2: codec.write_uint16(value) # U16 |
| when 3: codec.write_uint32(value) # U32 |
| when 4: codec.write_uint64(value) # U64 |
| when 6: codec.write_str8(value) # SSTR |
| when 7: codec.write_str16(value) # LSTR |
| when 8: codec.write_int64(value) # ABSTIME |
| when 9: codec.write_uint64(value) # DELTATIME |
| when 10: value.encode(codec) # REF |
| when 11: codec.write_uint8(value ? 1 : 0) # BOOL |
| when 12: codec.write_float(value) # FLOAT |
| when 13: codec.write_double(value) # DOUBLE |
| when 14: codec.write_uuid(value) # UUID |
| when 15: codec.write_map(value) # FTABLE |
| when 16: codec.write_int8(value) # S8 |
| when 17: codec.write_int16(value) # S16 |
| when 18: codec.write_int32(value) # S32 |
| when 19: codec.write_int64(value) # S64 |
| when 20: value.encode(codec) |
| when 21: # List |
| send_codec = Qpid::StringCodec.new(codec.spec) |
| encode_value(send_codec, value.size, 3) |
| value.each do v |
| ltype = encoding(v) |
| encode_value(send_codec,ltype,1) |
| encode_value(send_codec,v,ltype) |
| end |
| codec.write_vbin32(send_codec.encoded) |
| when 22: # Array |
| send_codec = Qpid::StringCodec.new(codec.spec) |
| encode_value(send_codec, value.size, 3) |
| if value.size > 0 |
| ltype = encoding(value[0]) |
| encode_value(send_codec,ltype,1) |
| value.each do v |
| encode_value(send_codec,v,ltype) |
| end |
| end |
| codec.write_vbin32(send_codec.encoded) |
| else |
| raise ValueError, "Invalid type code: %d" % typecode |
| end |
| end |
| |
| def display_value(value, typecode) |
| case typecode |
| when 1: return value.to_s |
| when 2: return value.to_s |
| when 3: return value.to_s |
| when 4: return value.to_s |
| when 6: return value.to_s |
| when 7: return value.to_s |
| when 8: return strftime("%c", gmtime(value / 1000000000)) |
| when 9: return value.to_s |
| when 10: return value.to_s |
| when 11: return value ? 'T' : 'F' |
| when 12: return value.to_s |
| when 13: return value.to_s |
| when 14: return Qpid::UUID::format(value) |
| when 15: return value.to_s |
| when 16: return value.to_s |
| when 17: return value.to_s |
| when 18: return value.to_s |
| when 19: return value.to_s |
| when 20: return value.to_s |
| when 21: return value.to_s |
| when 22: return value.to_s |
| else |
| raise ValueError, "Invalid type code: %d" % typecode |
| end |
| end |
| |
| private |
| |
| def binding_keys |
| key_list = [] |
| key_list << "schema.#" |
| if @rcv_objects && @rcv_events && @rcv_heartbeats && |
| ! @user_bindings |
| key_list << "console.#" |
| else |
| if @rcv_objects && ! @user_bindings |
| key_list << "console.obj.#" |
| else |
| key_list << "console.obj.*.*.org.apache.qpid.broker.agent" |
| end |
| key_list << "console.event.#" if @rcv_events |
| key_list << "console.heartbeat.#" if @rcv_heartbeats |
| end |
| return key_list |
| end |
| |
| # Check the object against select to check for a match |
| def select_match(object) |
| select.each do |key, value| |
| object.properties.each do |prop, propval| |
| return false if key == prop.name && value != propval |
| end |
| end |
| return true |
| end |
| |
| end |
| |
| class Package |
| attr_reader :name |
| |
| def initialize(name) |
| @name = name |
| end |
| end |
| |
| # A ClassKey uniquely identifies a class from the schema. |
| class ClassKey |
| attr_reader :package, :klass_name, :hash |
| |
| def initialize(package="", klass_name="", hash=0) |
| if (package.kind_of?(Qpid::Codec)) |
| @package = package.read_str8() |
| @klass_name = package.read_str8() |
| @hash = package.read_bin128() |
| else |
| @package = package |
| @klass_name = klass_name |
| @hash = hash |
| end |
| end |
| |
| def encode(codec) |
| codec.write_str8(@package) |
| codec.write_str8(@klass_name) |
| codec.write_bin128(@hash) |
| end |
| |
| def to_a() |
| return [@package, @klass_name, @hash] |
| end |
| |
| def hash_string() |
| "%08x-%08x-%08x-%08x" % hash.unpack("NNNN") |
| end |
| |
| def to_s() |
| return "#{@package}:#{@klass_name}(#{hash_string()})" |
| end |
| end |
| |
| class SchemaClass |
| |
| CLASS_KIND_TABLE = 1 |
| CLASS_KIND_EVENT = 2 |
| |
| attr_reader :klass_key, :arguments, :super_klass_key |
| |
| def initialize(session, kind, key, codec) |
| @session = session |
| @kind = kind |
| @klass_key = key |
| @super_klass_key = nil |
| @properties = [] |
| @statistics = [] |
| @methods = [] |
| @arguments = [] |
| |
| has_supertype = 0 #codec.read_uint8 |
| if @kind == CLASS_KIND_TABLE |
| prop_count = codec.read_uint16 |
| stat_count = codec.read_uint16 |
| method_count = codec.read_uint16 |
| if has_supertype == 1 |
| @super_klass_key = ClassKey.new(codec) |
| end |
| prop_count.times { |idx| |
| @properties << SchemaProperty.new(codec) } |
| stat_count.times { |idx| |
| @statistics << SchemaStatistic.new(codec) } |
| method_count.times { |idx| |
| @methods<< SchemaMethod.new(codec) } |
| elsif @kind == CLASS_KIND_EVENT |
| arg_count = codec.read_uint16 |
| arg_count.times { |idx| |
| sa = SchemaArgument.new(codec, false) |
| @arguments << sa |
| } |
| end |
| end |
| |
| def is_table? |
| @kind == CLASS_KIND_TABLE |
| end |
| |
| def is_event? |
| @kind == CLASS_KIND_EVENT |
| end |
| |
| def properties(include_inherited = true) |
| returnValue = @properties |
| if !@super_klass_key.nil? && include_inherited |
| returnValue = @properties + @session.schema(@super_klass_key).properties |
| end |
| return returnValue |
| end |
| |
| def statistics(include_inherited = true) |
| returnValue = @statistics |
| if !@super_klass_key.nil? && include_inherited |
| returnValue = @statistics + @session.schema(@super_klass_key).statistics |
| end |
| return returnValue |
| end |
| |
| def methods(include_inherited = true) |
| returnValue = @methods |
| if !@super_klass_key.nil? && include_inherited |
| returnValue = @methods + @session.schema(@super_klass_key).methods |
| end |
| return returnValue |
| end |
| |
| def to_s |
| if @kind == CLASS_KIND_TABLE |
| kind_str = "Table" |
| elsif @kind == CLASS_KIND_EVENT |
| kind_str = "Event" |
| else |
| kind_str = "Unsupported" |
| end |
| "#{kind_str} Class: #{klass_key.to_s}" |
| end |
| end |
| |
| class SchemaProperty |
| |
| attr_reader :name, :type, :access, :index, :optional, |
| :unit, :min, :max, :maxlen, :desc, :refClass, :refPackage |
| |
| def initialize(codec) |
| map = codec.read_map |
| @name = map["name"] |
| @type = map["type"] |
| @access = map["access"] |
| @index = map["index"] != 0 |
| @optional = map["optional"] != 0 |
| @unit = map["unit"] |
| @min = map["min"] |
| @max = map["max"] |
| @maxlen = map["maxlen"] |
| @desc = map["desc"] |
| @refClass = map["refClass"] |
| @refPackage = map["refPackage"] |
| end |
| |
| def to_s |
| @name |
| end |
| end |
| |
| class SchemaStatistic |
| |
| attr_reader :name, :type, :unit, :desc, :refClass, :refPackage |
| |
| def initialize(codec) |
| map = codec.read_map |
| @name = map["name"] |
| @type = map["type"] |
| @unit = map["unit"] |
| @desc = map["desc"] |
| @refClass = map["refClass"] |
| @refPackage = map["refPackage"] |
| end |
| |
| def to_s |
| @name |
| end |
| end |
| |
| class SchemaMethod |
| |
| attr_reader :name, :desc, :arguments |
| |
| def initialize(codec) |
| map = codec.read_map |
| @name = map["name"] |
| arg_count = map["argCount"] |
| @desc = map["desc"] |
| @arguments = [] |
| arg_count.times { |idx| |
| @arguments << SchemaArgument.new(codec, true) |
| } |
| end |
| |
| def to_s |
| result = @name + "(" |
| first = true |
| result += @arguments.select { |arg| arg.dir.index(?I) }.join(", ") |
| result += ")" |
| return result |
| end |
| end |
| |
| class SchemaArgument |
| |
| attr_reader :name, :type, :dir, :unit, :min, :max, :maxlen |
| attr_reader :desc, :default, :refClass, :refPackage |
| |
| def initialize(codec, method_arg) |
| map = codec.read_map |
| @name = map["name"] |
| @type = map["type"] |
| @dir = map["dir"].upcase if method_arg |
| @unit = map["unit"] |
| @min = map["min"] |
| @max = map["max"] |
| @maxlen = map["maxlen"] |
| @desc = map["desc"] |
| @default = map["default"] |
| @refClass = map["refClass"] |
| @refPackage = map["refPackage"] |
| end |
| end |
| |
| # Object that represents QMF object identifiers |
| class ObjectId |
| |
| include Comparable |
| |
| attr_reader :first, :second |
| |
| def initialize(codec, first=0, second=0) |
| if codec |
| @first = codec.read_uint64 |
| @second = codec.read_uint64 |
| else |
| @first = first |
| @second = second |
| end |
| end |
| |
| def <=>(other) |
| return 1 unless other.is_a?(ObjectId) |
| return -1 if first < other.first |
| return 1 if first > other.first |
| return second <=> other.second |
| end |
| |
| def to_s |
| "%d-%d-%d-%d-%d" % [flags, sequence, broker_bank, agent_bank, object] |
| end |
| |
| def index |
| [first, second] |
| end |
| |
| def flags |
| (first & 0xF000000000000000) >> 60 |
| end |
| |
| def sequence |
| (first & 0x0FFF000000000000) >> 48 |
| end |
| |
| def broker_bank |
| (first & 0x0000FFFFF0000000) >> 28 |
| end |
| |
| def agent_bank |
| first & 0x000000000FFFFFFF |
| end |
| |
| def object |
| second |
| end |
| |
| def durable? |
| sequence == 0 |
| end |
| |
| def encode(codec) |
| codec.write_uint64(first) |
| codec.write_uint64(second) |
| end |
| end |
| |
| class Object |
| |
| DEFAULT_METHOD_WAIT_TIME = 60 |
| |
| attr_reader :object_id, :schema, :properties, :statistics, |
| :current_time, :create_time, :delete_time, :broker |
| |
| def initialize(session, broker, schema, codec, prop, stat, managed=true) |
| @session = session |
| @broker = broker |
| @schema = schema |
| if managed |
| @current_time = codec.read_uint64 |
| @create_time = codec.read_uint64 |
| @delete_time = codec.read_uint64 |
| @object_id = ObjectId.new(codec) |
| end |
| @properties = [] |
| @statistics = [] |
| if prop |
| missing = parse_presence_masks(codec, schema) |
| schema.properties.each do |property| |
| v = nil |
| unless missing.include?(property.name) |
| v = @session.decode_value(codec, property.type) |
| end |
| @properties << [property, v] |
| end |
| end |
| |
| if stat |
| schema.statistics.each do |statistic| |
| s = @session.decode_value(codec, statistic.type) |
| @statistics << [statistic, s] |
| end |
| end |
| end |
| |
| def klass_key |
| @schema.klass_key |
| end |
| |
| |
| def methods |
| @schema.methods |
| end |
| |
| # Return the current, creation, and deletion times for this object |
| def timestamps |
| return [@current_time, @create_time, @delete_time] |
| end |
| |
| # Return a string describing this object's primary key |
| def index |
| @properties.select { |property, value| |
| property.index |
| }.collect { |property,value| |
| @session.display_value(value, property.type) }.join(":") |
| end |
| |
| # Replace properties and/or statistics with a newly received update |
| def merge_update(newer) |
| unless object_id == newer.object_id |
| raise "Objects with different object-ids" |
| end |
| @properties = newer.properties unless newer.properties.empty? |
| @statistics = newer.statistics unless newer.statistics.empty? |
| end |
| |
| def update |
| obj = @session.object(:object_id => @object_id, :broker => @broker) |
| if obj |
| merge_update(obj) |
| else |
| raise "Underlying object no longer exists." |
| end |
| end |
| |
| def to_s |
| @schema.klass_key.to_s |
| end |
| |
| # This must be defined because ruby has this (deprecated) method built in. |
| def id |
| method_missing(:id) |
| end |
| |
| # Same here.. |
| def type |
| method_missing(:type) |
| end |
| |
| def name |
| method_missing(:name) |
| end |
| |
| def method_missing(name, *args) |
| name = name.to_s |
| |
| if method = @schema.methods.find { |method| name == method.name } |
| return invoke(method, name, args) |
| end |
| |
| @properties.each do |property, value| |
| return value if name == property.name |
| if name == "_#{property.name}_" && property.type == 10 |
| # Dereference references |
| deref = @session.objects(:object_id => value, :broker => @broker) |
| return nil unless deref.size == 1 |
| return deref[0] |
| end |
| end |
| @statistics.each do |statistic, value| |
| if name == statistic.name |
| return value |
| end |
| end |
| raise "Type Object has no attribute '#{name}'" |
| end |
| |
| def encode(codec) |
| codec.write_uint8(20) |
| @schema.klass_key.encode(codec) |
| |
| # emit presence masks for optional properties |
| mask = 0 |
| bit = 0 |
| schema.properties.each do |property| |
| if prop.optional |
| bit = 1 if bit == 0 |
| mask |= bit if value |
| bit = bit << 1 |
| if bit == 256 |
| bit = 0 |
| codec.write_uint8(mask) |
| mask = 0 |
| end |
| codec.write_uint8(mask) if bit != 0 |
| end |
| end |
| |
| # encode properties |
| @properties.each do |property, value| |
| @session.encode_value(codec, value, prop.type) if value |
| end |
| |
| # encode statistics |
| @statistics.each do |statistic, value| |
| @session.encode_value(codec, value, stat.type) |
| end |
| end |
| |
| private |
| |
| def send_method_request(method, name, args, synchronous = false, time_wait = nil) |
| @schema.methods.each do |schema_method| |
| if name == schema_method.name |
| send_codec = Qpid::StringCodec.new(@broker.conn.spec) |
| seq = @session.seq_mgr.reserve([schema_method, synchronous]) |
| @broker.set_header(send_codec, ?M, seq) |
| @object_id.encode(send_codec) |
| @schema.klass_key.encode(send_codec) |
| send_codec.write_str8(name) |
| |
| formals = method.arguments.select { |arg| arg.dir.index(?I) } |
| count = method.arguments.select { |arg| arg.dir.index(?I) }.size |
| unless formals.size == args.size |
| raise "Incorrect number of arguments: expected #{formals.size}, got #{args.size}" |
| end |
| |
| formals.zip(args).each do |formal, actual| |
| @session.encode_value(send_codec, actual, formal.type) |
| end |
| |
| ttl = time_wait ? time_wait * 1000 : nil |
| smsg = @broker.message(send_codec.encoded, |
| "agent.#{object_id.broker_bank}.#{object_id.agent_bank}", ttl=ttl) |
| @broker.sync_start if synchronous |
| @broker.emit(smsg) |
| |
| return seq |
| end |
| end |
| end |
| |
| def invoke(method, name, args) |
| kwargs = args[args.size - 1] |
| sync = true |
| timeout = DEFAULT_METHOD_WAIT_TIME |
| |
| if kwargs.class == Hash |
| if kwargs.include?(:timeout) |
| timeout = kwargs[:timeout] |
| end |
| |
| if kwargs.include?(:async) |
| sync = !kwargs[:async] |
| end |
| args.pop |
| end |
| |
| seq = send_method_request(method, name, args, synchronous = sync) |
| if seq |
| return seq unless sync |
| unless @broker.wait_for_sync_done(timeout) |
| @session.seq_mgr.release(seq) |
| raise "Timed out waiting for method to respond" |
| end |
| |
| if @broker.error |
| error_text = @broker.error |
| @broker.error = nil |
| raise error_text |
| end |
| |
| return @broker.sync_result |
| end |
| raise "Invalid Method (software defect) [#{name}]" |
| end |
| |
| def parse_presence_masks(codec, schema) |
| exclude_list = [] |
| bit = 0 |
| schema.properties.each do |property| |
| if property.optional |
| if bit == 0 |
| mask = codec.read_uint8 |
| bit = 1 |
| end |
| if (mask & bit) == 0 |
| exclude_list << property.name |
| end |
| bit *= 2 |
| bit = 0 if bit == 256 |
| end |
| end |
| return exclude_list |
| end |
| end |
| |
| class MethodResult |
| |
| attr_reader :status, :text, :out_args |
| |
| def initialize(status, text, out_args) |
| @status = status |
| @text = text |
| @out_args = out_args |
| end |
| |
| def method_missing(name) |
| name = name.to_s() |
| if @out_args.include?(name) |
| return @out_args[name] |
| else |
| raise "Unknown method result arg #{name}" |
| end |
| end |
| |
| def to_s |
| argsString = "" |
| padding = "" |
| out_args.each do |key,value| |
| argsString += padding |
| padding = " " if padding == "" |
| argsString += key.to_s |
| argsString += " => " |
| argsString += value.to_s() |
| end |
| "MethodResult(Msg: '#{text}' Status: #{status} Return: [#{argsString}])" |
| end |
| end |
| |
| class ManagedConnection |
| |
| DELAY_MIN = 1 |
| DELAY_MAX = 128 |
| DELAY_FACTOR = 2 |
| include MonitorMixin |
| |
| def initialize(broker) |
| super() |
| @broker = broker |
| @cv = new_cond |
| @is_cancelled = false |
| end |
| |
| # Main body of the running thread. |
| def start |
| @thread = Thread.new { |
| delay = DELAY_MIN |
| while true |
| begin |
| @broker.try_to_connect |
| synchronize do |
| while !@is_cancelled and @broker.connected? |
| @cv.wait |
| Thread.exit if @is_cancelled |
| delay = DELAY_MIN |
| end |
| end |
| |
| rescue |
| delay *= DELAY_FACTOR if delay < DELAY_MAX |
| end |
| |
| synchronize do |
| @cv.wait(delay) |
| Thread.exit if @is_cancelled |
| end |
| end |
| } |
| end |
| |
| # Tell this thread to stop running and return. |
| def stop |
| synchronize do |
| @is_cancelled = true |
| @cv.signal |
| end |
| end |
| |
| # Notify the thread that the connection was lost. |
| def disconnected |
| synchronize do |
| @cv.signal |
| end |
| end |
| |
| def join |
| @thread.join |
| end |
| end |
| |
| class Broker |
| |
| SYNC_TIME = 60 |
| @@next_seq = 1 |
| |
| include MonitorMixin |
| |
| attr_accessor :error |
| |
| attr_reader :amqp_session_id, :amqp_session, :conn, :broker_bank, :topic_name |
| |
| attr_accessor :broker_id, :sync_result |
| |
| def initialize(session, host, port, auth_name, auth_pass, kwargs) |
| super() |
| |
| # For debugging.. |
| Thread.abort_on_exception = true |
| |
| @session = session |
| @host = host |
| @port = port |
| @auth_name = auth_name |
| @auth_pass = auth_pass |
| @user_id = nil |
| @auth_mechanism = kwargs[:mechanism] |
| @auth_service = kwargs[:service] |
| @broker_bank = 1 |
| @topic_bound = false |
| @cv = new_cond |
| @error = nil |
| @broker_id = nil |
| @is_connected = false |
| @amqp_session_id = "%s.%d.%d" % [Socket.gethostname, Process::pid, @@next_seq] |
| @@next_seq += 1 |
| @conn = nil |
| if @session.managedConnections? |
| @thread = ManagedConnection.new(self) |
| @thread.start |
| else |
| @thread = nil |
| try_to_connect |
| end |
| end |
| |
| def connected? |
| @is_connected |
| end |
| |
| def agent(broker_bank, agent_bank) |
| bank_key = "%d.%d" % [broker_bank, agent_bank] |
| return @agents[bank_key] |
| end |
| |
| # Get the list of agents reachable via this broker |
| def agents |
| @agents.values |
| end |
| |
| def url |
| "#{@host}:#{@port}" |
| end |
| |
| def to_s |
| if connected? |
| "Broker connected at: #{url}" |
| else |
| "Disconnected Broker" |
| end |
| end |
| |
| def wait_for_sync_done(timeout=nil) |
| wait_time = timeout ? timeout : SYNC_TIME |
| synchronize do |
| return @cv.wait_for(wait_time) { ! @sync_in_flight || @error } |
| end |
| end |
| |
| def wait_for_stable |
| synchronize do |
| return unless connected? |
| return if @reqs_outstanding == 0 |
| @sync_in_flight = true |
| unless @cv.wait_for(SYNC_TIME) { @reqs_outstanding == 0 } |
| raise "Timed out waiting for broker to synchronize" |
| end |
| end |
| end |
| |
| # Compose the header of a management message |
| def set_header(codec, opcode, seq=0) |
| codec.write_uint8(?A) |
| codec.write_uint8(?M) |
| codec.write_uint8(?2) |
| codec.write_uint8(opcode) |
| codec.write_uint32(seq) |
| end |
| |
| def message(body, routing_key="broker", ttl=nil) |
| dp = @amqp_session.delivery_properties |
| dp.routing_key = routing_key |
| dp.ttl = ttl if ttl |
| mp = @amqp_session.message_properties |
| mp.content_type = "x-application/qmf" |
| mp.reply_to = amqp_session.reply_to("amq.direct", @reply_name) |
| #mp.user_id = @user_id if @user_id |
| return Qpid::Message.new(dp, mp, body) |
| end |
| |
| def emit(msg, dest="qpid.management") |
| @amqp_session.message_transfer(:destination => dest, |
| :message => msg) |
| end |
| |
| def inc_outstanding |
| synchronize { @reqs_outstanding += 1 } |
| end |
| |
| def dec_outstanding |
| synchronize do |
| @reqs_outstanding -= 1 |
| if @reqs_outstanding == 0 && ! @topic_bound |
| @topic_bound = true |
| @session.binding_key_list.each do |key| |
| args = { |
| :exchange => "qpid.management", |
| :queue => @topic_name, |
| :binding_key => key } |
| @amqp_session.exchange_bind(args) |
| end |
| end |
| if @reqs_outstanding == 0 && @sync_in_flight |
| sync_done |
| end |
| end |
| end |
| |
| def sync_start |
| synchronize { @sync_in_flight = true } |
| end |
| |
| def sync_done |
| synchronize do |
| @sync_in_flight = false |
| @cv.signal |
| end |
| end |
| |
| def update_agent(obj) |
| bank_key = "%d.%d" % [obj.brokerBank, obj.agentBank] |
| if obj.delete_time == 0 |
| unless @agents.include?(bank_key) |
| agent = Agent.new(self, obj.agentBank, obj.label) |
| @agents[bank_key] = agent |
| @session.console.new_agent(agent) if @session.console |
| end |
| else |
| agent = @agents.delete(bank_key) |
| @session.console.del_agent(agent) if agent && @session.console |
| end |
| end |
| |
| def shutdown |
| if @thread |
| @thread.stop |
| @thread.join |
| end |
| if connected? |
| @amqp_session.incoming("rdest").stop |
| if @session.console |
| @amqp_session.incoming("tdest").stop |
| end |
| @amqp_session.close |
| @is_connected = false |
| end |
| end |
| |
| def try_to_connect |
| @agents = {} |
| @agents["1.0"] = Agent.new(self, 0, "BrokerAgent") |
| @topic_bound = false |
| @sync_in_flight = false |
| @sync_request = 0 |
| @sync_result = nil |
| @reqs_outstanding = 1 |
| |
| # FIXME: Need sth for Qpid::Util::connect |
| |
| @conn = Qpid::Connection.new(TCPSocket.new(@host, @port), |
| :mechanism => @auth_mechanism, |
| :username => @auth_name, |
| :password => @auth_pass, |
| :host => @host, |
| :service => @auth_service) |
| @conn.start |
| @user_id = @conn.user_id |
| @reply_name = "reply-%s" % amqp_session_id |
| @amqp_session = @conn.session(@amqp_session_id) |
| @amqp_session.auto_sync = true |
| |
| @amqp_session.queue_declare(:queue => @reply_name, |
| :exclusive => true, |
| :auto_delete => true) |
| |
| @amqp_session.exchange_bind(:exchange => "amq.direct", |
| :queue => @reply_name, |
| :binding_key => @reply_name) |
| @amqp_session.message_subscribe(:queue => @reply_name, |
| :destination => "rdest", |
| :accept_mode => @amqp_session.message_accept_mode.none, |
| :acquire_mode => @amqp_session.message_acquire_mode.pre_acquired) |
| q = @amqp_session.incoming("rdest") |
| q.exc_listen(& method(:exception_cb)) |
| q.listen(& method(:reply_cb)) |
| @amqp_session.message_set_flow_mode(:destination => "rdest", |
| :flow_mode => 1) |
| @amqp_session.message_flow(:destination => "rdest", |
| :unit => 0, |
| :value => 0xFFFFFFFF) |
| @amqp_session.message_flow(:destination => "rdest", |
| :unit => 1, |
| :value => 0xFFFFFFFF) |
| |
| @topic_name = "topic-#{@amqp_session_id}" |
| @amqp_session.queue_declare(:queue => @topic_name, |
| :exclusive => true, |
| :auto_delete => true) |
| @amqp_session.message_subscribe(:queue => @topic_name, |
| :destination => "tdest", |
| :accept_mode => @amqp_session.message_accept_mode.none, |
| :acquire_mode => @amqp_session.message_acquire_mode.pre_acquired) |
| @amqp_session.incoming("tdest").listen(& method(:reply_cb)) |
| @amqp_session.message_set_flow_mode(:destination => "tdest", |
| :flow_mode => 1) |
| @amqp_session.message_flow(:destination => "tdest", |
| :unit => 0, |
| :value => 0xFFFFFFFF) |
| @amqp_session.message_flow(:destination => "tdest", |
| :unit => 1, |
| :value => 0xFFFFFFFF) |
| |
| @is_connected = true |
| @session.handle_broker_connect(self) |
| |
| codec = Qpid::StringCodec.new(@conn.spec) |
| set_header(codec, ?B) |
| msg = message(codec.encoded) |
| emit(msg) |
| end |
| |
| private |
| |
| # Check the header of a management message and extract the opcode and |
| # class |
| def check_header(codec) |
| begin |
| return [nil, nil] unless codec.read_uint8 == ?A |
| return [nil, nil] unless codec.read_uint8 == ?M |
| return [nil, nil] unless codec.read_uint8 == ?2 |
| opcode = codec.read_uint8 |
| seq = codec.read_uint32 |
| return [opcode, seq] |
| rescue |
| return [nil, nil] |
| end |
| end |
| |
| def reply_cb(msg) |
| codec = Qpid::StringCodec.new(@conn.spec, msg.body) |
| loop do |
| opcode, seq = check_header(codec) |
| return unless opcode |
| case opcode |
| when ?b: @session.handle_broker_resp(self, codec, seq) |
| when ?p: @session.handle_package_ind(self, codec, seq) |
| when ?z: @session.handle_command_complete(self, codec, seq) |
| when ?q: @session.handle_class_ind(self, codec, seq) |
| when ?m: @session.handle_method_resp(self, codec, seq) |
| when ?h: @session.handle_heartbeat_ind(self, codec, seq, msg) |
| when ?e: @session.handle_event_ind(self, codec, seq) |
| when ?s: @session.handle_schema_resp(self, codec, seq) |
| when ?c: @session.handle_content_ind(self, codec, seq, true, false) |
| when ?i: @session.handle_content_ind(self, codec, seq, false, true) |
| when ?g: @session.handle_content_ind(self, codec, seq, true, true) |
| else |
| raise "Unexpected opcode #{opcode.inspect}" |
| end |
| end |
| end |
| |
| def exception_cb(data) |
| @is_connected = false |
| @error = data |
| synchronize { @cv.signal if @sync_in_flight } |
| @session.handle_error(@error) |
| @session.handle_broker_disconnect(self) |
| @thread.disconnected if @thread |
| end |
| end |
| |
| class Agent |
| attr_reader :broker, :agent_bank, :label |
| |
| def initialize(broker, agent_bank, label) |
| @broker = broker |
| @agent_bank = agent_bank |
| @label = label |
| end |
| |
| def broker_bank |
| @broker.broker_bank |
| end |
| |
| def to_s |
| "Agent at bank %d.%d (%s)" % [@broker.broker_bank, @agent_bank, @label] |
| end |
| end |
| |
| class Event |
| |
| attr_reader :klass_key, :arguments, :timestamp, :name, :schema |
| |
| def initialize(session, broker, codec) |
| @session = session |
| @broker = broker |
| @klass_key = ClassKey.new(codec) |
| @timestamp = codec.read_int64 |
| @severity = codec.read_uint8 |
| @schema = nil |
| |
| pname, cname, hash = @klass_key.to_a() |
| session.packages.keys.each do |pname| |
| k = [cname, hash] |
| if session.packages[pname].include?(k) |
| @schema = session.packages[pname][k] |
| @arguments = {} |
| @schema.arguments.each do |arg| |
| v = session.decode_value(codec, arg.type) |
| @arguments[arg.name] = v |
| end |
| end |
| end |
| end |
| |
| def to_s |
| return "<uninterpretable>" unless @schema |
| t = Time.at(self.timestamp / 1000000000) |
| out = t.strftime("%c") |
| out += " " + sev_name + " " + @klass_key.package + ":" + @klass_key.klass_name |
| out += " broker=" + @broker.url |
| @schema.arguments.each do |arg| |
| out += " " + arg.name + "=" + @session.display_value(@arguments[arg.name], arg.type) |
| end |
| return out |
| end |
| |
| def sev_name |
| case @severity |
| when 0 : return "EMER " |
| when 1 : return "ALERT" |
| when 2 : return "CRIT " |
| when 3 : return "ERROR" |
| when 4 : return "WARN " |
| when 5 : return "NOTIC" |
| when 6 : return "INFO " |
| when 7 : return "DEBUG" |
| else |
| return "INV-%d" % @severity |
| end |
| end |
| |
| end |
| |
| # Manage sequence numbers for asynchronous method calls |
| class SequenceManager |
| include MonitorMixin |
| |
| def initialize |
| super() |
| @sequence = 0 |
| @pending = {} |
| end |
| |
| # Reserve a unique sequence number |
| def reserve (data) |
| synchronize do |
| result = @sequence |
| @sequence += 1 |
| @pending[result] = data |
| return result |
| end |
| end |
| |
| # Release a reserved sequence number |
| def release (seq) |
| synchronize { @pending.delete(seq) } |
| end |
| end |
| |
| class DebugConsole < Console |
| |
| def broker_connected(broker) |
| puts "brokerConnected #{broker}" |
| end |
| |
| def broker_disconnected(broker) |
| puts "brokerDisconnected #{broker}" |
| end |
| |
| def new_package(name) |
| puts "newPackage #{name}" |
| end |
| |
| def new_class(kind, klass_key) |
| puts "newClass #{kind} #{klass_key}" |
| end |
| |
| def new_agent(agent) |
| puts "new_agent #{agent}" |
| end |
| |
| def del_agent(agent) |
| puts "delAgent #{agent}" |
| end |
| |
| def object_props(broker, record) |
| puts "objectProps #{record}" |
| end |
| |
| def object_stats(broker, record) |
| puts "objectStats #{record}" |
| end |
| |
| def event(broker, event) |
| puts "event #{event}" |
| end |
| |
| def heartbeat(agent, timestamp) |
| puts "heartbeat #{agent}" |
| end |
| |
| def broker_info(broker) |
| puts "brokerInfo #{broker}" |
| end |
| end |
| |
| module XML |
| TYPES = { |
| 1 => "uint8", |
| 2 => "uint16", |
| 3 => "uint32", |
| 4 => "uint64", |
| 5 => "bool", |
| 6 => "short-stirng", |
| 7 => "long-string", |
| 8 => "abs-time", |
| 9 => "delta-time", |
| 10 => "reference", |
| 11 => "boolean", |
| 12 => "float", |
| 13 => "double", |
| 14 => "uuid", |
| 15 => "field-table", |
| 16 => "int8", |
| 17 => "int16", |
| 18 => "int32", |
| 19 => "int64", |
| 20 => "object", |
| 21 => "list", |
| 22 => "array" |
| } |
| |
| ACCESS_MODES = { |
| 1 => "RC", |
| 2 => "RW", |
| 3 => "RO" |
| } |
| |
| def common_attributes(item) |
| attr_string = "" |
| attr_string << " desc='#{item.desc}'" if item.desc |
| attr_string << " desc='#{item.desc}'" if item.desc |
| attr_string << " refPackage='#{item.refPackage}'" if item.refPackage |
| attr_string << " refClass='#{item.refClass}'" if item.refClass |
| attr_string << " unit='#{item.unit}'" if item.unit |
| attr_string << " min='#{item.min}'" if item.min |
| attr_string << " max='#{item.max}'" if item.max |
| attr_string << " maxlen='#{item.maxlen}'" if item.maxlen |
| return attr_string |
| end |
| |
| module_function :common_attributes |
| |
| def schema_xml(session, *packages) |
| schema = "<schemas>\n" |
| packages.each do |package| |
| schema << "\t<schema package='#{package}'>\n" |
| session.classes(package).each do |klass_key| |
| klass = session.schema(klass_key) |
| if klass.is_table? |
| if klass.super_klass_key |
| schema << "\t\t<class name='#{klass.klass_key.klass_name}' hash='#{klass.klass_key.hash_string}' extends='#{klass.super_klass_key.to_s}'>\n" |
| else |
| schema << "\t\t<class name='#{klass.klass_key.klass_name}' hash='#{klass.klass_key.hash_string}'>\n" |
| end |
| klass.properties(false).each do |property| |
| schema << "\t\t\t<property name='#{property.name}' type='#{TYPES[property.type]}' access='#{ACCESS_MODES[property.access]}' optional='#{property.optional ? "True" : "False"}'#{common_attributes(property)}/>\n" |
| end |
| klass.methods(false).each do |method| |
| schema << "\t\t\t<method name='#{method.name}'>\n" |
| method.arguments.each do |arg| |
| schema << "\t\t\t\t<arg name='#{arg.name}' dir='#{arg.dir}' type='#{TYPES[arg.type]}'#{common_attributes(arg)}/>\n" |
| end |
| schema << "\t\t\t</method>\n" |
| end |
| schema << "\t\t</class>\n" |
| else |
| schema << "\t\t<event name='#{klass.klass_key.klass_name}' hash='#{klass.klass_key.hash_string}'>\n" |
| klass.arguments.each do |arg| |
| schema << "\t\t\t<arg name='#{arg.name}'type='#{TYPES[arg.type]}'#{common_attributes(arg)}/>\n" |
| end |
| schema << "\t\t</event>\n" |
| end |
| end |
| schema << "\t</package>\n" |
| end |
| schema << "</schema>" |
| end |
| |
| module_function :schema_xml |
| end |
| |
| end |
