| # |
| # 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. |
| # |
| import sys |
| import socket |
| import os |
| import logging |
| from threading import Thread |
| from threading import RLock |
| from threading import Condition |
| import qmfengine |
| from qmfengine import (ACCESS_READ_CREATE, ACCESS_READ_ONLY, ACCESS_READ_WRITE) |
| from qmfengine import (CLASS_EVENT, CLASS_OBJECT) |
| from qmfengine import (DIR_IN, DIR_IN_OUT, DIR_OUT) |
| from qmfengine import (TYPE_ABSTIME, TYPE_ARRAY, TYPE_BOOL, TYPE_DELTATIME, |
| TYPE_DOUBLE, TYPE_FLOAT, TYPE_INT16, TYPE_INT32, TYPE_INT64, |
| TYPE_INT8, TYPE_LIST, TYPE_LSTR, TYPE_MAP, TYPE_OBJECT, |
| TYPE_REF, TYPE_SSTR, TYPE_UINT16, TYPE_UINT32, TYPE_UINT64, |
| TYPE_UINT8, TYPE_UUID) |
| from qmfengine import (O_EQ, O_NE, O_LT, O_LE, O_GT, O_GE, O_RE_MATCH, O_RE_NOMATCH, |
| E_NOT, E_AND, E_OR, E_XOR) |
| from qmfengine import (SEV_EMERG, SEV_ALERT, SEV_CRIT, SEV_ERROR, SEV_WARN, SEV_NOTICE, |
| SEV_INFORM, SEV_DEBUG) |
| |
| |
| def qmf_to_native(val): |
| typecode = val.getType() |
| if typecode == TYPE_UINT8: return val.asUint() |
| elif typecode == TYPE_UINT16: return val.asUint() |
| elif typecode == TYPE_UINT32: return val.asUint() |
| elif typecode == TYPE_UINT64: return val.asUint64() |
| elif typecode == TYPE_SSTR: return val.asString() |
| elif typecode == TYPE_LSTR: return val.asString() |
| elif typecode == TYPE_ABSTIME: return val.asInt64() |
| elif typecode == TYPE_DELTATIME: return val.asUint64() |
| elif typecode == TYPE_REF: return ObjectId(val.asObjectId()) |
| elif typecode == TYPE_BOOL: return val.asBool() |
| elif typecode == TYPE_FLOAT: return val.asFloat() |
| elif typecode == TYPE_DOUBLE: return val.asDouble() |
| elif typecode == TYPE_UUID: return val.asUuid() |
| elif typecode == TYPE_INT8: return val.asInt() |
| elif typecode == TYPE_INT16: return val.asInt() |
| elif typecode == TYPE_INT32: return val.asInt() |
| elif typecode == TYPE_INT64: return val.asInt64() |
| elif typecode == TYPE_MAP: return value_to_dict(val) |
| elif typecode == TYPE_LIST: return value_to_list(val) |
| else: |
| # when TYPE_OBJECT |
| logging.error( "Unsupported type for get_attr? '%s'" % str(val.getType()) ) |
| return None |
| |
| |
| def native_to_qmf(target, value): |
| val = None |
| typecode = None |
| if target.__class__ == qmfengine.Value: |
| val = target |
| typecode = val.getType() |
| else: |
| typecode = target |
| val = qmfengine.Value(typecode) |
| |
| if typecode == TYPE_UINT8: val.setUint(value) |
| elif typecode == TYPE_UINT16: val.setUint(value) |
| elif typecode == TYPE_UINT32: val.setUint(value) |
| elif typecode == TYPE_UINT64: val.setUint64(value) |
| elif typecode == TYPE_SSTR: |
| if value: val.setString(value) |
| else: val.setString('') |
| elif typecode == TYPE_LSTR: |
| if value: val.setString(value) |
| else: val.setString('') |
| elif typecode == TYPE_ABSTIME: val.setInt64(value) |
| elif typecode == TYPE_DELTATIME: val.setUint64(value) |
| elif typecode == TYPE_REF: val.setObjectId(value.impl) |
| elif typecode == TYPE_BOOL: val.setBool(value) |
| elif typecode == TYPE_FLOAT: val.setFloat(value) |
| elif typecode == TYPE_DOUBLE: val.setDouble(value) |
| elif typecode == TYPE_UUID: val.setUuid(value) |
| elif typecode == TYPE_INT8: val.setInt(value) |
| elif typecode == TYPE_INT16: val.setInt(value) |
| elif typecode == TYPE_INT32: val.setInt(value) |
| elif typecode == TYPE_INT64: val.setInt64(value) |
| elif typecode == TYPE_MAP: dict_to_value(val, value) |
| elif typecode == TYPE_LIST: list_to_value(val, value) |
| else: |
| # when TYPE_OBJECT |
| logging.error("Unsupported type for get_attr? '%s'" % str(val.getType())) |
| return None |
| return val |
| |
| |
| def pick_qmf_type(value): |
| if value.__class__ == int: |
| if value >= 0: |
| if value < 0x100000000: return TYPE_UINT32 |
| return TYPE_UINT64 |
| else: |
| if value > -0xffffffff: return TYPE_INT32 |
| return TYPE_INT64 |
| |
| if value.__class__ == long: |
| if value >= 0: return TYPE_UINT64 |
| return TYPE_INT64 |
| |
| if value.__class__ == str: |
| if len(value) < 256: return TYPE_SSTR |
| return TYPE_LSTR |
| |
| if value.__class__ == float: return TYPE_DOUBLE |
| if value.__class__ == bool: return TYPE_BOOL |
| if value == None: return TYPE_BOOL |
| if value.__class__ == dict: return TYPE_MAP |
| if value.__class__ == list: return TYPE_LIST |
| |
| raise "QMF type not known for native type %s" % value.__class__ |
| |
| |
| def value_to_dict(val): |
| if not val.isMap(): raise "value_to_dict must be given a map value" |
| mymap = {} |
| for i in range(val.keyCount()): |
| key = val.key(i) |
| mymap[key] = qmf_to_native(val.byKey(key)) |
| return mymap |
| |
| |
| def dict_to_value(val, mymap): |
| for key, value in mymap.items(): |
| if key.__class__ != str: raise "QMF map key must be a string" |
| typecode = pick_qmf_type(value) |
| val.insert(key, native_to_qmf(typecode, value)) |
| |
| |
| def value_to_list(val): |
| mylist = [] |
| if val.isList(): |
| for i in range(val.listItemCount()): |
| mylist.append(qmf_to_native(val.listItem(i))) |
| return mylist |
| #if val.isArray(): |
| # for i in range(val.arrayItemCount()): |
| # mylist.append(qmf_to_native(val.arrayItem(i))) |
| # return mylist |
| |
| raise "value_to_list must be given a list value" |
| |
| |
| def list_to_value(val, mylist): |
| for item in mylist: |
| typecode = pick_qmf_type(item) |
| val.appendToList(native_to_qmf(typecode, item)) |
| |
| |
| ##============================================================================== |
| ## CONNECTION |
| ##============================================================================== |
| |
| class ConnectionSettings(object): |
| #attr_reader :impl |
| def __init__(self, url=None): |
| if url: |
| self.impl = qmfengine.ConnectionSettings(url) |
| else: |
| self.impl = qmfengine.ConnectionSettings() |
| |
| |
| def set_attr(self, key, val): |
| if type(val) == str: |
| _v = qmfengine.Value(TYPE_LSTR) |
| _v.setString(val) |
| elif type(val) == int: |
| _v = qmfengine.Value(TYPE_UINT32) |
| _v.setUint(val) |
| elif type(val) == bool: |
| _v = qmfengine.Value(TYPE_BOOL) |
| _v.setBool(val) |
| else: |
| raise Exception("Argument error: value for attribute '%s' has unsupported type: %s" % ( key, type(val))) |
| |
| good = self.impl.setAttr(key, _v) |
| if not good: |
| raise Exception("Argument error: unsupported attribute '%s'" % key ) |
| |
| |
| def get_attr(self, key): |
| _v = self.impl.getAttr(key) |
| if _v.isString(): |
| return _v.asString() |
| elif _v.isUint(): |
| return _v.asUint() |
| elif _v.isBool(): |
| return _v.asBool() |
| else: |
| raise Exception("Argument error: value for attribute '%s' has unsupported type: %s" % ( key, str(_v.getType()))) |
| |
| |
| def __getattr__(self, name): |
| return self.get_attr(name) |
| |
| |
| def __setattr__(self, name, value): |
| if name == "impl": |
| return super.__setattr__(self, name, value) |
| return self.set_attr(name, value) |
| |
| |
| |
| class ConnectionHandler: |
| def conn_event_connected(self): None |
| def conn_event_disconnected(self, error): None |
| def conn_event_visit(self): None |
| def sess_event_session_closed(self, context, error): None |
| def sess_event_recv(self, context, message): None |
| |
| |
| |
| class Connection(Thread): |
| def __init__(self, settings): |
| Thread.__init__(self) |
| self._lock = RLock() |
| self.impl = qmfengine.ResilientConnection(settings.impl) |
| self._sockEngine, self._sock = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM) |
| self.impl.setNotifyFd(self._sockEngine.fileno()) |
| self._new_conn_handlers = [] |
| self._conn_handlers_to_delete = [] |
| self._conn_handlers = [] |
| self._connected = False |
| self._operational = True |
| self.start() |
| |
| |
| def destroy(self, timeout=None): |
| logging.debug("Destroying Connection...") |
| self._operational = False |
| self.kick() |
| self.join(timeout) |
| logging.debug("... Conn Destroyed!" ) |
| if self.isAlive(): |
| logging.error("Error: Connection thread '%s' is hung..." % self.getName()) |
| |
| |
| def connected(self): |
| return self._connected |
| |
| |
| def kick(self): |
| self.impl.notify() |
| |
| |
| def add_conn_handler(self, handler): |
| self._lock.acquire() |
| try: |
| self._new_conn_handlers.append(handler) |
| finally: |
| self._lock.release() |
| self.kick() |
| |
| |
| def del_conn_handler(self, handler): |
| self._lock.acquire() |
| try: |
| self._conn_handlers_to_delete.append(handler) |
| finally: |
| self._lock.release() |
| self.kick() |
| |
| |
| def run(self): |
| eventImpl = qmfengine.ResilientConnectionEvent() |
| new_handlers = [] |
| del_handlers = [] |
| bt_count = 0 |
| |
| while self._operational: |
| logging.debug("Connection thread waiting for socket data...") |
| self._sock.recv(1) |
| |
| self._lock.acquire() |
| try: |
| new_handlers = self._new_conn_handlers |
| del_handlers = self._conn_handlers_to_delete |
| self._new_conn_handlers = [] |
| self._conn_handlers_to_delete = [] |
| finally: |
| self._lock.release() |
| |
| for nh in new_handlers: |
| self._conn_handlers.append(nh) |
| if self._connected: |
| nh.conn_event_connected() |
| new_handlers = [] |
| |
| for dh in del_handlers: |
| if dh in self._conn_handlers: |
| self._conn_handlers.remove(dh) |
| del_handlers = [] |
| |
| valid = self.impl.getEvent(eventImpl) |
| while valid: |
| try: |
| if eventImpl.kind == qmfengine.ResilientConnectionEvent.CONNECTED: |
| logging.debug("Connection thread: CONNECTED event received.") |
| self._connected = True |
| for h in self._conn_handlers: |
| h.conn_event_connected() |
| |
| elif eventImpl.kind == qmfengine.ResilientConnectionEvent.DISCONNECTED: |
| logging.debug("Connection thread: DISCONNECTED event received.") |
| self._connected = False |
| for h in self._conn_handlers: |
| h.conn_event_disconnected(eventImpl.errorText) |
| |
| elif eventImpl.kind == qmfengine.ResilientConnectionEvent.SESSION_CLOSED: |
| logging.debug("Connection thread: SESSION_CLOSED event received.") |
| eventImpl.sessionContext.handler.sess_event_session_closed(eventImpl.sessionContext, eventImpl.errorText) |
| |
| elif eventImpl.kind == qmfengine.ResilientConnectionEvent.RECV: |
| logging.debug("Connection thread: RECV event received.") |
| eventImpl.sessionContext.handler.sess_event_recv(eventImpl.sessionContext, eventImpl.message) |
| else: |
| logging.debug("Connection thread received unknown event: '%s'" % str(eventImpl.kind)) |
| |
| except: |
| import traceback |
| logging.error( "Exception occurred during Connection event processing:" ) |
| logging.error( str(sys.exc_info()) ) |
| if bt_count < 2: |
| traceback.print_exc() |
| traceback.print_stack() |
| bt_count += 1 |
| |
| self.impl.popEvent() |
| valid = self.impl.getEvent(eventImpl) |
| |
| for h in self._conn_handlers: |
| h.conn_event_visit() |
| |
| logging.debug("Shutting down Connection thread") |
| |
| |
| |
| class Session: |
| def __init__(self, conn, label, handler): |
| self._conn = conn |
| self._label = label |
| self.handler = handler |
| self.handle = qmfengine.SessionHandle() |
| result = self._conn.impl.createSession(label, self, self.handle) |
| |
| |
| def destroy(self): |
| self._conn.impl.destroySession(self.handle) |
| |
| |
| |
| ##============================================================================== |
| ## OBJECTS and EVENTS |
| ##============================================================================== |
| |
| class QmfEvent(object): |
| # attr_reader :impl, :event_class |
| def __init__(self, cls, kwargs={}): |
| self._allow_sets = True |
| if kwargs.has_key("broker"): |
| self._broker = kwargs["broker"] |
| else: |
| self._broker = None |
| if cls: |
| self.event_class = cls |
| self.impl = qmfengine.Event(self.event_class.impl) |
| elif kwargs.has_key("impl"): |
| self.impl = qmfengine.Event(kwargs["impl"]) |
| self.event_class = SchemaEventClass(None, None, 0, |
| {"impl":self.impl.getClass()}) |
| else: |
| raise Exception("Argument error: required parameter ('impl') not supplied") |
| |
| |
| def arguments(self): |
| list = [] |
| for arg in self.event_class.arguments: |
| list.append([arg, self.get_attr(arg.name())]) |
| return list |
| |
| |
| def get_attr(self, name): |
| val = self._value(name) |
| return qmf_to_native(val) |
| |
| |
| def set_attr(self, name, v): |
| val = self._value(name) |
| native_to_qmf(val, v) |
| |
| |
| def __getitem__(self, name): |
| return self.get_attr(name) |
| |
| |
| def __setitem__(self, name, value): |
| self.set_attr(name, value) |
| |
| |
| def __setattr__(self, name, value): |
| # |
| # Ignore the internal attributes, set them normally... |
| # |
| if (name[0] == '_' or |
| name == 'impl' or |
| name == 'event_class'): |
| return super.__setattr__(self, name, value) |
| |
| if not self._allow_sets: |
| raise Exception("'Set' operations not permitted on this object") |
| |
| # |
| # If the name matches an argument name, set the value of the argument. |
| # |
| # print "set name=%s" % str(name) |
| for arg in self.event_class.arguments: |
| if arg.name() == name: |
| return self.set_attr(name, value) |
| |
| # unrecognized name? should I raise an exception? |
| super.__setattr__(self, name, value) |
| |
| |
| def __getattr__(self, name, *args): |
| # |
| # If the name matches an argument name, return the value of the argument. |
| # |
| for arg in self.event_class.arguments: |
| if arg.name() == name: |
| return self.get_attr(name) |
| |
| # |
| # This name means nothing to us, pass it up the line to the parent |
| # class's handler. |
| # |
| # print "__getattr__=%s" % str(name) |
| super.__getattr__(self, name) |
| |
| |
| def _value(self, name): |
| val = self.impl.getValue(name) |
| if not val: |
| raise Exception("Argument error: attribute named '%s' not defined for package %s, class %s" % |
| (name, |
| self.event_class.impl.getClassKey().getPackageName(), |
| self.event_class.impl.getClassKey().getClassName())) |
| return val |
| |
| |
| class QmfObject(object): |
| # attr_reader :impl, :object_class |
| def __init__(self, cls, kwargs={}): |
| self._cv = Condition() |
| self._sync_count = 0 |
| self._sync_result = None |
| self._allow_sets = False |
| if kwargs.has_key("broker"): |
| self._broker = kwargs["broker"] |
| else: |
| self._broker = None |
| if cls: |
| self.object_class = cls |
| self.impl = qmfengine.Object(self.object_class.impl) |
| elif kwargs.has_key("impl"): |
| self.impl = qmfengine.Object(kwargs["impl"]) |
| self.object_class = SchemaObjectClass(None, |
| None, |
| {"impl":self.impl.getClass()}) |
| else: |
| raise Exception("Argument error: required parameter ('impl') not supplied") |
| |
| |
| def destroy(self): |
| self.impl.destroy() |
| |
| |
| def object_id(self): |
| return ObjectId(self.impl.getObjectId()) |
| |
| |
| def set_object_id(self, oid): |
| self.impl.setObjectId(oid.impl) |
| |
| |
| def properties(self): |
| list = [] |
| for prop in self.object_class.properties: |
| list.append([prop, self.get_attr(prop.name())]) |
| return list |
| |
| |
| def statistics(self): |
| list = [] |
| for stat in self.object_class.statistics: |
| list.append([stat, self.get_attr(stat.name())]) |
| return list |
| |
| |
| def get_attr(self, name): |
| val = self._value(name) |
| return qmf_to_native(val) |
| |
| |
| def set_attr(self, name, v): |
| val = self._value(name) |
| native_to_qmf(val, v) |
| |
| |
| def __getitem__(self, name): |
| return self.get_attr(name) |
| |
| |
| def __setitem__(self, name, value): |
| self.set_attr(name, value) |
| |
| |
| def inc_attr(self, name, by=1): |
| self.set_attr(name, self.get_attr(name) + by) |
| |
| |
| def dec_attr(self, name, by=1): |
| self.set_attr(name, self.get_attr(name) - by) |
| |
| |
| def __setattr__(self, name, value): |
| # |
| # Ignore the internal attributes, set them normally... |
| # |
| if (name[0] == '_' or |
| name == 'impl' or |
| name == 'object_class'): |
| return super.__setattr__(self, name, value) |
| |
| if not self._allow_sets: |
| raise Exception("'Set' operations not permitted on this object") |
| # |
| # If the name matches a property name, set the value of the property. |
| # |
| # print "set name=%s" % str(name) |
| for prop in self.object_class.properties: |
| if prop.name() == name: |
| return self.set_attr(name, value) |
| # |
| # otherwise, check for a statistic set... |
| # |
| for stat in self.object_class.statistics: |
| if stat.name() == name: |
| return self.set_attr(name, value) |
| |
| # unrecognized name? should I raise an exception? |
| super.__setattr__(self, name, value) |
| |
| |
| def __getattr__(self, name, *args): |
| # |
| # If the name matches a property name, return the value of the property. |
| # |
| for prop in self.object_class.properties: |
| if prop.name() == name: |
| return self.get_attr(name) |
| # |
| # Do the same for statistics |
| # |
| for stat in self.object_class.statistics: |
| if stat.name() == name: |
| return self.get_attr(name) |
| # |
| # If we still haven't found a match for the name, check to see if |
| # it matches a method name. If so, marshall up the arguments into |
| # a map, and invoke the method. |
| # |
| for method in self.object_class.methods: |
| if method.name() == name: |
| argMap = self._marshall(method, args) |
| return lambda name, argMap : self._invokeMethod(name, argMap) |
| |
| # |
| # This name means nothing to us, pass it up the line to the parent |
| # class's handler. |
| # |
| # print "__getattr__=%s" % str(name) |
| super.__getattr__(self, name) |
| |
| |
| def _invokeMethod(self, name, argMap): |
| """ |
| Private: Helper function that invokes an object's method, and waits for the result. |
| """ |
| self._cv.acquire() |
| try: |
| timeout = 30 |
| self._sync_count = 1 |
| self.impl.invokeMethod(name, argMap, self) |
| if self._broker: |
| self._broker.conn.kick() |
| self._cv.wait(timeout) |
| if self._sync_count == 1: |
| raise Exception("Timed out: waiting for response to method call.") |
| finally: |
| self._cv.release() |
| |
| return self._sync_result |
| |
| |
| def _method_result(self, result): |
| """ |
| Called to return the result of a method call on an object |
| """ |
| self._cv.acquire(); |
| try: |
| self._sync_result = result |
| self._sync_count -= 1 |
| self._cv.notify() |
| finally: |
| self._cv.release() |
| |
| |
| def _marshall(schema, args): |
| ''' |
| Private: Convert a list of arguments (positional) into a Value object of type "map". |
| Used to create the argument parameter for an object's method invokation. |
| ''' |
| # Build a map of the method's arguments |
| map = qmfengine.Value(TYPE_MAP) |
| for arg in schema.arguments: |
| if arg.direction == DIR_IN or arg.direction == DIR_IN_OUT: |
| map.insert(arg.name, qmfengine.Value(arg.typecode)) |
| |
| # install each argument's value into the map |
| marshalled = Arguments(map) |
| idx = 0 |
| for arg in schema.arguments: |
| if arg.direction == DIR_IN or arg.direction == DIR_IN_OUT: |
| if args[idx]: |
| marshalled[arg.name] = args[idx] |
| idx += 1 |
| |
| return marshalled.map |
| |
| |
| def _value(self, name): |
| val = self.impl.getValue(name) |
| if not val: |
| raise Exception("Argument error: attribute named '%s' not defined for package %s, class %s" % |
| (name, |
| self.object_class.impl.getClassKey().getPackageName(), |
| self.object_class.impl.getClassKey().getClassName())) |
| return val |
| |
| |
| |
| class AgentObject(QmfObject): |
| def __init__(self, cls, kwargs={}): |
| QmfObject.__init__(self, cls, kwargs) |
| self._allow_sets = True |
| |
| |
| def destroy(self): |
| self.impl.destroy() |
| |
| |
| def set_object_id(self, oid): |
| self.impl.setObjectId(oid.impl) |
| |
| |
| |
| class ConsoleObject(QmfObject): |
| # attr_reader :current_time, :create_time, :delete_time |
| def __init__(self, cls, kwargs={}): |
| QmfObject.__init__(self, cls, kwargs) |
| |
| |
| def update(self): |
| if not self._broker: |
| raise Exception("No linkage to broker") |
| newer = self._broker.console.objects(Query({"object_id":object_id})) |
| if newer.size != 1: |
| raise Exception("Expected exactly one update for this object, %d present" % int(newer.size)) |
| self.merge_update(newer[0]) |
| |
| |
| def merge_update(self, newObject): |
| self.impl.merge(new_object.impl) |
| |
| |
| def is_deleted(self): |
| return self.impl.isDeleted() |
| |
| |
| def key(self): pass |
| |
| |
| |
| class ObjectId: |
| def __init__(self, impl=None): |
| if impl: |
| self.impl = impl |
| else: |
| self.impl = qmfengine.ObjectId() |
| self.agent_key = "%d.%d" % (self.impl.getBrokerBank(), self.impl.getAgentBank()) |
| |
| |
| def object_num_high(self): |
| return self.impl.getObjectNumHi() |
| |
| |
| def object_num_low(self): |
| return self.impl.getObjectNumLo() |
| |
| |
| def agent_key(self): |
| self.agent_key |
| |
| def __eq__(self, other): |
| if not isinstance(other, self.__class__): return False |
| return self.impl == other.impl |
| |
| def __ne__(self, other): |
| return not self.__eq__(other) |
| |
| def __repr__(self): |
| return self.impl.str() |
| |
| |
| |
| class Arguments(object): |
| def __init__(self, map): |
| self.map = map |
| self._by_hash = {} |
| key_count = self.map.keyCount() |
| a = 0 |
| while a < key_count: |
| key = self.map.key(a) |
| self._by_hash[key] = qmf_to_native(self.map.byKey(key)) |
| a += 1 |
| |
| |
| def __getitem__(self, key): |
| return self._by_hash[key] |
| |
| |
| def __setitem__(self, key, value): |
| self._by_hash[key] = value |
| self.set(key, value) |
| |
| |
| def __iter__(self): |
| return self._by_hash.__iter__ |
| |
| |
| def __getattr__(self, name): |
| if name in self._by_hash: |
| return self._by_hash[name] |
| return super.__getattr__(self, name) |
| |
| |
| def __setattr__(self, name, value): |
| # |
| # ignore local data members |
| # |
| if (name[0] == '_' or |
| name == 'map'): |
| return super.__setattr__(self, name, value) |
| |
| if name in self._by_hash: |
| self._by_hash[name] = value |
| return self.set(name, value) |
| |
| return super.__setattr__(self, name, value) |
| |
| |
| def set(self, key, value): |
| val = self.map.byKey(key) |
| native_to_qmf(val, value) |
| |
| |
| |
| class MethodResponse(object): |
| def __init__(self, impl): |
| self.impl = qmfengine.MethodResponse(impl) |
| |
| |
| def status(self): |
| return self.impl.getStatus() |
| |
| |
| def exception(self): |
| return self.impl.getException() |
| |
| |
| def text(self): |
| return exception().asString() |
| |
| |
| def args(self): |
| return Arguments(self.impl.getArgs()) |
| |
| |
| def __getattr__(self, name): |
| myArgs = self.args() |
| return myArgs.__getattr__(name) |
| |
| |
| def __setattr__(self, name, value): |
| if name == 'impl': |
| return super.__setattr__(self, name, value) |
| |
| myArgs = self.args() |
| return myArgs.__setattr__(name, value) |
| |
| |
| |
| ##============================================================================== |
| ## QUERY |
| ##============================================================================== |
| |
| |
| class Query: |
| def __init__(self, kwargs={}): |
| if "impl" in kwargs: |
| self.impl = kwargs["impl"] |
| else: |
| package = '' |
| if "key" in kwargs: |
| # construct using SchemaClassKey: |
| self.impl = qmfengine.Query(kwargs["key"]) |
| elif "object_id" in kwargs: |
| self.impl = qmfengine.Query(kwargs["object_id"].impl) |
| else: |
| if "package" in kwargs: |
| package = kwargs["package"] |
| if "class" in kwargs: |
| self.impl = qmfengine.Query(kwargs["class"], package) |
| else: |
| raise Exception("Argument error: invalid arguments, use 'key', 'object_id' or 'class'[,'package']") |
| |
| |
| def package_name(self): return self.impl.getPackage() |
| def class_name(self): return self.impl.getClass() |
| def object_id(self): |
| _objid = self.impl.getObjectId() |
| if _objid: |
| return ObjectId(_objid) |
| else: |
| return None |
| |
| |
| ##============================================================================== |
| ## SCHEMA |
| ##============================================================================== |
| |
| |
| |
| class SchemaArgument: |
| #attr_reader :impl |
| def __init__(self, name, typecode, kwargs={}): |
| if "impl" in kwargs: |
| self.impl = kwargs["impl"] |
| else: |
| self.impl = qmfengine.SchemaArgument(name, typecode) |
| if kwargs.has_key("dir"): self.impl.setDirection(kwargs["dir"]) |
| if kwargs.has_key("unit"): self.impl.setUnit(kwargs["unit"]) |
| if kwargs.has_key("desc"): self.impl.setDesc(kwargs["desc"]) |
| |
| |
| def name(self): |
| return self.impl.getName() |
| |
| |
| def direction(self): |
| return self.impl.getDirection() |
| |
| |
| def typecode(self): |
| return self.impl.getType() |
| |
| |
| def __repr__(self): |
| return self.name() |
| |
| |
| |
| class SchemaMethod: |
| # attr_reader :impl, arguments |
| def __init__(self, name, kwargs={}): |
| self.arguments = [] |
| if "impl" in kwargs: |
| self.impl = kwargs["impl"] |
| for i in range(self.impl.getArgumentCount()): |
| self.arguments.append(SchemaArgument(None,None,{"impl":self.impl.getArgument(i)})) |
| else: |
| self.impl = qmfengine.SchemaMethod(name) |
| if kwargs.has_key("desc"): self.impl.setDesc(kwargs["desc"]) |
| |
| |
| def add_argument(self, arg): |
| self.arguments.append(arg) |
| self.impl.addArgument(arg.impl) |
| |
| def name(self): |
| return self.impl.getName() |
| |
| def __repr__(self): |
| return self.name() |
| |
| |
| |
| class SchemaProperty: |
| #attr_reader :impl |
| def __init__(self, name, typecode, kwargs={}): |
| if "impl" in kwargs: |
| self.impl = kwargs["impl"] |
| else: |
| self.impl = qmfengine.SchemaProperty(name, typecode) |
| if kwargs.has_key("access"): self.impl.setAccess(kwargs["access"]) |
| if kwargs.has_key("index"): self.impl.setIndex(kwargs["index"]) |
| if kwargs.has_key("optional"): self.impl.setOptional(kwargs["optional"]) |
| if kwargs.has_key("unit"): self.impl.setUnit(kwargs["unit"]) |
| if kwargs.has_key("desc"): self.impl.setDesc(kwargs["desc"]) |
| |
| |
| def name(self): |
| return self.impl.getName() |
| |
| def __repr__(self): |
| return self.name() |
| |
| |
| |
| class SchemaStatistic: |
| # attr_reader :impl |
| def __init__(self, name, typecode, kwargs={}): |
| if "impl" in kwargs: |
| self.impl = kwargs["impl"] |
| else: |
| self.impl = qmfengine.SchemaStatistic(name, typecode) |
| if kwargs.has_key("unit"): self.impl.setUnit(kwargs["unit"]) |
| if kwargs.has_key("desc"): self.impl.setDesc(kwargs["desc"]) |
| |
| |
| def name(self): |
| return self.impl.getName() |
| |
| def __repr__(self): |
| return self.name() |
| |
| |
| |
| class SchemaClassKey: |
| #attr_reader :impl |
| def __init__(self, i): |
| self.impl = i |
| |
| |
| def package_name(self): |
| return self.impl.getPackageName() |
| |
| |
| def class_name(self): |
| return self.impl.getClassName() |
| |
| def __repr__(self): |
| return self.impl.asString() |
| |
| |
| |
| class SchemaObjectClass: |
| # attr_reader :impl, :properties, :statistics, :methods |
| def __init__(self, package, name, kwargs={}): |
| self.properties = [] |
| self.statistics = [] |
| self.methods = [] |
| if "impl" in kwargs: |
| self.impl = kwargs["impl"] |
| |
| for i in range(self.impl.getPropertyCount()): |
| self.properties.append(SchemaProperty(None, None, {"impl":self.impl.getProperty(i)})) |
| |
| for i in range(self.impl.getStatisticCount()): |
| self.statistics.append(SchemaStatistic(None, None, {"impl":self.impl.getStatistic(i)})) |
| |
| for i in range(self.impl.getMethodCount()): |
| self.methods.append(SchemaMethod(None, {"impl":self.impl.getMethod(i)})) |
| else: |
| self.impl = qmfengine.SchemaObjectClass(package, name) |
| |
| |
| def add_property(self, prop): |
| self.properties.append(prop) |
| self.impl.addProperty(prop.impl) |
| |
| |
| def add_statistic(self, stat): |
| self.statistics.append(stat) |
| self.impl.addStatistic(stat.impl) |
| |
| |
| def add_method(self, meth): |
| self.methods.append(meth) |
| self.impl.addMethod(meth.impl) |
| |
| |
| def class_key(self): |
| return SchemaClassKey(self.impl.getClassKey()) |
| |
| |
| def package_name(self): |
| return self.impl.getClassKey().getPackageName() |
| |
| |
| def class_name(self): |
| return self.impl.getClassKey().getClassName() |
| |
| |
| |
| class SchemaEventClass: |
| # attr_reader :impl :arguments |
| def __init__(self, package, name, sev, kwargs={}): |
| self.arguments = [] |
| if "impl" in kwargs: |
| self.impl = kwargs["impl"] |
| for i in range(self.impl.getArgumentCount()): |
| self.arguments.append(SchemaArgument(nil, nil, {"impl":self.impl.getArgument(i)})) |
| else: |
| self.impl = qmfengine.SchemaEventClass(package, name, sev) |
| if kwargs.has_key("desc"): self.impl.setDesc(kwargs["desc"]) |
| |
| |
| def add_argument(self, arg): |
| self.arguments.append(arg) |
| self.impl.addArgument(arg.impl) |
| |
| |
| def name(self): |
| return self.impl.getClassKey().getClassName() |
| |
| def class_key(self): |
| return SchemaClassKey(self.impl.getClassKey()) |
| |
| |
| def package_name(self): |
| return self.impl.getClassKey().getPackageName() |
| |
| |
| def class_name(self): |
| return self.impl.getClassKey().getClassName() |
| |
| |
| ##============================================================================== |
| ## CONSOLE |
| ##============================================================================== |
| |
| |
| |
| class ConsoleHandler: |
| def agent_added(self, agent): pass |
| def agent_deleted(self, agent): pass |
| def new_package(self, package): pass |
| def new_class(self, class_key): pass |
| def object_update(self, obj, hasProps, hasStats): pass |
| def event_received(self, event): pass |
| def agent_heartbeat(self, agent, timestamp): pass |
| def method_response(self, resp): pass |
| def broker_info(self, broker): pass |
| |
| |
| |
| class Console(Thread): |
| # attr_reader :impl |
| def __init__(self, handler=None, kwargs={}): |
| Thread.__init__(self) |
| self._handler = handler |
| self.impl = qmfengine.Console() |
| self._event = qmfengine.ConsoleEvent() |
| self._broker_list = [] |
| self._cv = Condition() |
| self._sync_count = 0 |
| self._sync_result = None |
| self._select = {} |
| self._cb_cond = Condition() |
| self._operational = True |
| self.start() |
| |
| |
| def destroy(self, timeout=None): |
| logging.debug("Destroying Console...") |
| self._operational = False |
| self.start_console_events() # wake thread up |
| self.join(timeout) |
| logging.debug("... Console Destroyed!") |
| if self.isAlive(): |
| logging.error( "Console thread '%s' is hung..." % self.getName() ) |
| |
| |
| def add_connection(self, conn): |
| broker = Broker(self, conn) |
| self._cv.acquire() |
| try: |
| self._broker_list.append(broker) |
| finally: |
| self._cv.release() |
| return broker |
| |
| |
| def del_connection(self, broker): |
| logging.debug("shutting down broker...") |
| broker.shutdown() |
| logging.debug("...broker down.") |
| self._cv.acquire() |
| try: |
| self._broker_list.remove(broker) |
| finally: |
| self._cv.release() |
| logging.debug("del_connection() finished") |
| |
| |
| def packages(self): |
| plist = [] |
| for i in range(self.impl.packageCount()): |
| plist.append(self.impl.getPackageName(i)) |
| return plist |
| |
| |
| def classes(self, package, kind=CLASS_OBJECT): |
| clist = [] |
| for i in range(self.impl.classCount(package)): |
| key = self.impl.getClass(package, i) |
| class_kind = self.impl.getClassKind(key) |
| if class_kind == kind: |
| if kind == CLASS_OBJECT: |
| clist.append(SchemaObjectClass(None, None, {"impl":self.impl.getObjectClass(key)})) |
| elif kind == CLASS_EVENT: |
| clist.append(SchemaEventClass(None, None, 0, {"impl":self.impl.getEventClass(key)})) |
| return clist |
| |
| |
| def bind_package(self, package): |
| return self.impl.bindPackage(package) |
| |
| |
| def bind_class(self, kwargs = {}): |
| if "key" in kwargs: |
| self.impl.bindClass(kwargs["key"]) |
| elif "package" in kwargs: |
| package = kwargs["package"] |
| if "class" in kwargs: |
| self.impl.bindClass(package, kwargs["class"]) |
| else: |
| self.impl.bindClass(package, "*") |
| else: |
| raise Exception("Argument error: invalid arguments, use 'key' or 'package'[,'class']") |
| |
| |
| def bind_event(self, kwargs = {}): |
| if "key" in kwargs: |
| self.impl.bindEvent(kwargs["key"]) |
| elif "package" in kwargs: |
| package = kwargs["package"] |
| if "event" in kwargs: |
| self.impl.bindEvent(package, kwargs["event"]) |
| else: |
| self.impl.bindEvent(package, "*") |
| else: |
| raise Exception("Argument error: invalid arguments, use 'key' or 'package'[,'event']") |
| |
| |
| def agents(self, broker=None): |
| blist = [] |
| if broker: |
| blist.append(broker) |
| else: |
| self._cv.acquire() |
| try: |
| # copy while holding lock |
| blist = self._broker_list[:] |
| finally: |
| self._cv.release() |
| |
| agents = [] |
| for b in blist: |
| for idx in range(b.impl.agentCount()): |
| agents.append(AgentProxy(b.impl.getAgent(idx), b)) |
| |
| return agents |
| |
| |
| def objects(self, query, kwargs = {}): |
| timeout = 30 |
| agent = None |
| temp_args = kwargs.copy() |
| if type(query) == type({}): |
| temp_args.update(query) |
| |
| if "_timeout" in temp_args: |
| timeout = temp_args["_timeout"] |
| temp_args.pop("_timeout") |
| |
| if "_agent" in temp_args: |
| agent = temp_args["_agent"] |
| temp_args.pop("_agent") |
| |
| if type(query) == type({}): |
| query = Query(temp_args) |
| |
| self._select = {} |
| for k in temp_args.iterkeys(): |
| if type(k) == str: |
| self._select[k] = temp_args[k] |
| |
| self._cv.acquire() |
| try: |
| self._sync_count = 1 |
| self._sync_result = [] |
| broker = self._broker_list[0] |
| broker.send_query(query.impl, None, agent) |
| self._cv.wait(timeout) |
| if self._sync_count == 1: |
| raise Exception("Timed out: waiting for query response") |
| finally: |
| self._cv.release() |
| |
| return self._sync_result |
| |
| |
| def object(self, query, kwargs = {}): |
| ''' |
| Return one and only one object or None. |
| ''' |
| objs = objects(query, kwargs) |
| if len(objs) == 1: |
| return objs[0] |
| else: |
| return None |
| |
| |
| def first_object(self, query, kwargs = {}): |
| ''' |
| Return the first of potentially many objects. |
| ''' |
| objs = objects(query, kwargs) |
| if objs: |
| return objs[0] |
| else: |
| return None |
| |
| |
| # Check the object against select to check for a match |
| def _select_match(self, object): |
| schema_props = object.properties() |
| for key in self._select.iterkeys(): |
| for prop in schema_props: |
| if key == p[0].name() and self._select[key] != p[1]: |
| return False |
| return True |
| |
| |
| def _get_result(self, list, context): |
| ''' |
| Called by Broker proxy to return the result of a query. |
| ''' |
| self._cv.acquire() |
| try: |
| for item in list: |
| if self._select_match(item): |
| self._sync_result.append(item) |
| self._sync_count -= 1 |
| self._cv.notify() |
| finally: |
| self._cv.release() |
| |
| |
| def start_sync(self, query): pass |
| |
| |
| def touch_sync(self, sync): pass |
| |
| |
| def end_sync(self, sync): pass |
| |
| |
| def run(self): |
| while self._operational: |
| self._cb_cond.acquire() |
| try: |
| self._cb_cond.wait(1) |
| while self._do_console_events(): |
| pass |
| finally: |
| self._cb_cond.release() |
| logging.debug("Shutting down Console thread") |
| |
| |
| def start_console_events(self): |
| self._cb_cond.acquire() |
| try: |
| self._cb_cond.notify() |
| finally: |
| self._cb_cond.release() |
| |
| |
| def _do_console_events(self): |
| ''' |
| Called by the Console thread to poll for events. Passes the events |
| onto the ConsoleHandler associated with this Console. Is called |
| periodically, but can also be kicked by Console.start_console_events(). |
| ''' |
| count = 0 |
| valid = self.impl.getEvent(self._event) |
| while valid: |
| count += 1 |
| try: |
| if self._event.kind == qmfengine.ConsoleEvent.AGENT_ADDED: |
| logging.debug("Console Event AGENT_ADDED received") |
| if self._handler: |
| self._handler.agent_added(AgentProxy(self._event.agent, None)) |
| elif self._event.kind == qmfengine.ConsoleEvent.AGENT_DELETED: |
| logging.debug("Console Event AGENT_DELETED received") |
| if self._handler: |
| self._handler.agent_deleted(AgentProxy(self._event.agent, None)) |
| elif self._event.kind == qmfengine.ConsoleEvent.NEW_PACKAGE: |
| logging.debug("Console Event NEW_PACKAGE received") |
| if self._handler: |
| self._handler.new_package(self._event.name) |
| elif self._event.kind == qmfengine.ConsoleEvent.NEW_CLASS: |
| logging.debug("Console Event NEW_CLASS received") |
| if self._handler: |
| self._handler.new_class(SchemaClassKey(self._event.classKey)) |
| elif self._event.kind == qmfengine.ConsoleEvent.OBJECT_UPDATE: |
| logging.debug("Console Event OBJECT_UPDATE received") |
| if self._handler: |
| self._handler.object_update(ConsoleObject(None, {"impl":self._event.object}), |
| self._event.hasProps, self._event.hasStats) |
| elif self._event.kind == qmfengine.ConsoleEvent.EVENT_RECEIVED: |
| logging.debug("Console Event EVENT_RECEIVED received") |
| elif self._event.kind == qmfengine.ConsoleEvent.AGENT_HEARTBEAT: |
| logging.debug("Console Event AGENT_HEARTBEAT received") |
| if self._handler: |
| self._handler.agent_heartbeat(AgentProxy(self._event.agent, None), self._event.timestamp) |
| elif self._event.kind == qmfengine.ConsoleEvent.METHOD_RESPONSE: |
| logging.debug("Console Event METHOD_RESPONSE received") |
| else: |
| logging.debug("Console thread received unknown event: '%s'" % str(self._event.kind)) |
| except e: |
| print "Exception caught in callback thread:", e |
| self.impl.popEvent() |
| valid = self.impl.getEvent(self._event) |
| return count |
| |
| |
| |
| class AgentProxy: |
| # attr_reader :broker |
| def __init__(self, impl, broker): |
| self.impl = impl |
| self.broker = broker |
| self.key = "%d.%d" % (self.impl.getBrokerBank(), self.impl.getAgentBank()) |
| |
| |
| def label(self): |
| return self.impl.getLabel() |
| |
| |
| def key(self): |
| return self.key |
| |
| |
| class Broker(ConnectionHandler): |
| # attr_reader :impl :conn, :console, :broker_bank |
| def __init__(self, console, conn): |
| self.broker_bank = 1 |
| self.console = console |
| self.conn = conn |
| self._session = None |
| self._cv = Condition() |
| self._stable = None |
| self._event = qmfengine.BrokerEvent() |
| self._xmtMessage = qmfengine.Message() |
| self.impl = qmfengine.BrokerProxy(self.console.impl) |
| self.console.impl.addConnection(self.impl, self) |
| self.conn.add_conn_handler(self) |
| self._operational = True |
| |
| |
| def shutdown(self): |
| logging.debug("broker.shutdown() called.") |
| self.console.impl.delConnection(self.impl) |
| self.conn.del_conn_handler(self) |
| if self._session: |
| self.impl.sessionClosed() |
| logging.debug("broker.shutdown() sessionClosed done.") |
| self._session.destroy() |
| logging.debug("broker.shutdown() session destroy done.") |
| self._session = None |
| self._operational = False |
| logging.debug("broker.shutdown() done.") |
| |
| |
| def wait_for_stable(self, timeout = None): |
| self._cv.acquire() |
| try: |
| if self._stable: |
| return |
| if timeout: |
| self._cv.wait(timeout) |
| if not self._stable: |
| raise Exception("Timed out: waiting for broker connection to become stable") |
| else: |
| while not self._stable: |
| self._cv.wait() |
| finally: |
| self._cv.release() |
| |
| |
| def send_query(self, query, ctx, agent): |
| agent_impl = None |
| if agent: |
| agent_impl = agent.impl |
| self.impl.sendQuery(query, ctx, agent_impl) |
| self.conn.kick() |
| |
| |
| def _do_broker_events(self): |
| count = 0 |
| valid = self.impl.getEvent(self._event) |
| while valid: |
| count += 1 |
| if self._event.kind == qmfengine.BrokerEvent.BROKER_INFO: |
| logging.debug("Broker Event BROKER_INFO received"); |
| elif self._event.kind == qmfengine.BrokerEvent.DECLARE_QUEUE: |
| logging.debug("Broker Event DECLARE_QUEUE received"); |
| self.conn.impl.declareQueue(self._session.handle, self._event.name) |
| elif self._event.kind == qmfengine.BrokerEvent.DELETE_QUEUE: |
| logging.debug("Broker Event DELETE_QUEUE received"); |
| self.conn.impl.deleteQueue(self._session.handle, self._event.name) |
| elif self._event.kind == qmfengine.BrokerEvent.BIND: |
| logging.debug("Broker Event BIND received"); |
| self.conn.impl.bind(self._session.handle, self._event.exchange, self._event.name, self._event.bindingKey) |
| elif self._event.kind == qmfengine.BrokerEvent.UNBIND: |
| logging.debug("Broker Event UNBIND received"); |
| self.conn.impl.unbind(self._session.handle, self._event.exchange, self._event.name, self._event.bindingKey) |
| elif self._event.kind == qmfengine.BrokerEvent.SETUP_COMPLETE: |
| logging.debug("Broker Event SETUP_COMPLETE received"); |
| self.impl.startProtocol() |
| elif self._event.kind == qmfengine.BrokerEvent.STABLE: |
| logging.debug("Broker Event STABLE received"); |
| self._cv.acquire() |
| try: |
| self._stable = True |
| self._cv.notify() |
| finally: |
| self._cv.release() |
| elif self._event.kind == qmfengine.BrokerEvent.QUERY_COMPLETE: |
| result = [] |
| for idx in range(self._event.queryResponse.getObjectCount()): |
| result.append(ConsoleObject(None, {"impl":self._event.queryResponse.getObject(idx), "broker":self})) |
| self.console._get_result(result, self._event.context) |
| elif self._event.kind == qmfengine.BrokerEvent.METHOD_RESPONSE: |
| obj = self._event.context |
| obj._method_result(MethodResponse(self._event.methodResponse())) |
| |
| self.impl.popEvent() |
| valid = self.impl.getEvent(self._event) |
| |
| return count |
| |
| |
| def _do_broker_messages(self): |
| count = 0 |
| valid = self.impl.getXmtMessage(self._xmtMessage) |
| while valid: |
| count += 1 |
| logging.debug("Broker: sending msg on connection") |
| self.conn.impl.sendMessage(self._session.handle, self._xmtMessage) |
| self.impl.popXmt() |
| valid = self.impl.getXmtMessage(self._xmtMessage) |
| |
| return count |
| |
| |
| def _do_events(self): |
| while True: |
| self.console.start_console_events() |
| bcnt = self._do_broker_events() |
| mcnt = self._do_broker_messages() |
| if bcnt == 0 and mcnt == 0: |
| break; |
| |
| |
| def conn_event_connected(self): |
| logging.debug("Broker: Connection event CONNECTED") |
| self._session = Session(self.conn, "qmfc-%s.%d" % (socket.gethostname(), os.getpid()), self) |
| self.impl.sessionOpened(self._session.handle) |
| self._do_events() |
| |
| |
| def conn_event_disconnected(self, error): |
| logging.debug("Broker: Connection event DISCONNECTED") |
| pass |
| |
| |
| def conn_event_visit(self): |
| self._do_events() |
| |
| |
| def sess_event_session_closed(self, context, error): |
| logging.debug("Broker: Session event CLOSED") |
| self.impl.sessionClosed() |
| |
| |
| def sess_event_recv(self, context, message): |
| logging.debug("Broker: Session event MSG_RECV") |
| if not self._operational: |
| logging.warning("Unexpected session event message received by Broker proxy: context='%s'" % str(context)) |
| self.impl.handleRcvMessage(message) |
| self._do_events() |
| |
| |
| |
| ##============================================================================== |
| ## AGENT |
| ##============================================================================== |
| |
| |
| |
| class AgentHandler: |
| def get_query(self, context, query, userId): None |
| def method_call(self, context, name, object_id, args, userId): None |
| |
| |
| |
| class Agent(ConnectionHandler): |
| def __init__(self, handler, label=""): |
| if label == "": |
| self._agentLabel = "rb-%s.%d" % (socket.gethostname(), os.getpid()) |
| else: |
| self._agentLabel = label |
| self._conn = None |
| self._handler = handler |
| self.impl = qmfengine.Agent(self._agentLabel) |
| self._event = qmfengine.AgentEvent() |
| self._xmtMessage = qmfengine.Message() |
| |
| |
| def set_connection(self, conn): |
| self._conn = conn |
| self._conn.add_conn_handler(self) |
| |
| |
| def register_class(self, cls): |
| self.impl.registerClass(cls.impl) |
| |
| |
| def alloc_object_id(self, low = 0, high = 0): |
| return ObjectId(self.impl.allocObjectId(low, high)) |
| |
| |
| def raise_event(self, event): |
| self.impl.raiseEvent(event.impl) |
| |
| def query_response(self, context, obj): |
| self.impl.queryResponse(context, obj.impl) |
| |
| |
| def query_complete(self, context): |
| self.impl.queryComplete(context) |
| |
| |
| def method_response(self, context, status, text, arguments): |
| self.impl.methodResponse(context, status, text, arguments.map) |
| |
| |
| def do_agent_events(self): |
| count = 0 |
| valid = self.impl.getEvent(self._event) |
| while valid: |
| count += 1 |
| if self._event.kind == qmfengine.AgentEvent.GET_QUERY: |
| self._handler.get_query(self._event.sequence, |
| Query({"impl":self._event.query}), |
| self._event.authUserId) |
| |
| elif self._event.kind == qmfengine.AgentEvent.START_SYNC: |
| pass |
| elif self._event.kind == qmfengine.AgentEvent.END_SYNC: |
| pass |
| elif self._event.kind == qmfengine.AgentEvent.METHOD_CALL: |
| args = Arguments(self._event.arguments) |
| self._handler.method_call(self._event.sequence, self._event.name, |
| ObjectId(self._event.objectId), |
| args, self._event.authUserId) |
| |
| elif self._event.kind == qmfengine.AgentEvent.DECLARE_QUEUE: |
| self._conn.impl.declareQueue(self._session.handle, self._event.name) |
| |
| elif self._event.kind == qmfengine.AgentEvent.DELETE_QUEUE: |
| self._conn.impl.deleteQueue(self._session.handle, self._event.name) |
| |
| elif self._event.kind == qmfengine.AgentEvent.BIND: |
| self._conn.impl.bind(self._session.handle, self._event.exchange, |
| self._event.name, self._event.bindingKey) |
| |
| elif self._event.kind == qmfengine.AgentEvent.UNBIND: |
| self._conn.impl.unbind(self._session.handle, self._event.exchange, |
| self._event.name, self._event.bindingKey) |
| |
| elif self._event.kind == qmfengine.AgentEvent.SETUP_COMPLETE: |
| self.impl.startProtocol() |
| |
| self.impl.popEvent() |
| valid = self.impl.getEvent(self._event) |
| return count |
| |
| |
| def do_agent_messages(self): |
| count = 0 |
| valid = self.impl.getXmtMessage(self._xmtMessage) |
| while valid: |
| count += 1 |
| self._conn.impl.sendMessage(self._session.handle, self._xmtMessage) |
| self.impl.popXmt() |
| valid = self.impl.getXmtMessage(self._xmtMessage) |
| return count |
| |
| |
| def do_events(self): |
| while True: |
| ecnt = self.do_agent_events() |
| mcnt = self.do_agent_messages() |
| if ecnt == 0 and mcnt == 0: |
| break |
| |
| |
| def conn_event_connected(self): |
| logging.debug("Agent Connection Established...") |
| self._session = Session(self._conn, |
| "qmfa-%s.%d" % (socket.gethostname(), os.getpid()), |
| self) |
| self.impl.newSession() |
| self.do_events() |
| |
| |
| def conn_event_disconnected(self, error): |
| logging.debug("Agent Connection Lost") |
| pass |
| |
| |
| def conn_event_visit(self): |
| self.do_events() |
| |
| |
| def sess_event_session_closed(self, context, error): |
| logging.debug("Agent Session Lost") |
| pass |
| |
| |
| def sess_event_recv(self, context, message): |
| self.impl.handleRcvMessage(message) |
| self.do_events() |
| |
| |
