# -*- Mode: Python; -*- # Package : omniORBpy # tcInternal.py Created on: 1999/06/24 # Author : Duncan Grisby (dpg1) # # Copyright (C) 1999 AT&T Laboratories Cambridge # # This file is part of the omniORBpy library # # The omniORBpy library is free software; you can redistribute it # and/or modify it under the terms of the GNU Lesser General # Public License as published by the Free Software Foundation; # either version 2.1 of the License, or (at your option) any later # version. # # This library is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Lesser General Public License for more details. # # You should have received a copy of the GNU Lesser General Public # License along with this library; if not, write to the Free # Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, # MA 02111-1307, USA # # # Description: # TypeCode internal implementation # $Id: tcInternal.py,v 1.9.2.3 2001/05/14 14:21:15 dpg1 Exp $ # $Log: tcInternal.py,v $ # Revision 1.9.2.3 2001/05/14 14:21:15 dpg1 # TypeCode.equivalent() was broken in a similar manner to # get_compact_typecode() # # Revision 1.9.2.2 2001/05/11 16:27:23 dpg1 # TypeCode.get_compact_typecode() was broken. # # Revision 1.9.2.1 2000/08/07 09:19:24 dpg1 # Long long support # # Revision 1.9 2000/01/31 10:51:41 dpg1 # Fix to exception throwing. # # Revision 1.8 1999/11/02 10:38:31 dpg1 # Last bug wasn't quite fixed. # # Revision 1.7 1999/11/01 20:59:42 dpg1 # Fixed bug in insertIndirections() if the same node is indirected to # more than once. # # Revision 1.6 1999/09/24 09:22:00 dpg1 # Added copyright notices. # # Revision 1.5 1999/09/20 15:11:45 dpg1 # Bug in insertIndirections() fixed. # # Revision 1.4 1999/09/13 14:52:27 dpg1 # TypeCode equivalence. # # Revision 1.3 1999/07/29 14:17:18 dpg1 # TypeCode creation interface. # # Revision 1.2 1999/07/19 15:48:40 dpg1 # All sorts of fixes. # # Revision 1.1 1999/06/24 15:23:28 dpg1 # Initial revision # import omniORB, CORBA import types # The TypeCode implementation is based upon descriptors consisting of # Python tuples, as used by the marshalling code. Although the public # interface to TypeCodes presents a graph of connected TypeCode # objects, the graph is actually only fully-stored by the descriptor # tuple. A root TypeCode object has a reference to the top of the # descriptor tuple. Non-root TypeCode objects are only created when # requested. # # Recursive descriptors pose a problem for Python's garbage collector. # To ensure they are properly collected, all non-root TypeCode objects # keep a reference to the root TypeCode object. The root TypeCode is # therefore only deleted once all "views" of the TypeCode have gone. # When the root TypeCode object is deleted, it recursively descends # its descriptor, and removes any indirections it finds. All loops are # thus removed from the descriptor, making it a candidate for # collection. # # This approach means that the descriptor becomes invalid once the # root TypeCode object is deleted, even if there are other active # references to the descriptor. BEWARE! # Kinds as numbers: tv_null = 0 tv_void = 1 tv_short = 2 tv_long = 3 tv_ushort = 4 tv_ulong = 5 tv_float = 6 tv_double = 7 tv_boolean = 8 tv_char = 9 tv_octet = 10 tv_any = 11 tv_TypeCode = 12 tv_Principal = 13 tv_objref = 14 tv_struct = 15 tv_union = 16 tv_enum = 17 tv_string = 18 tv_sequence = 19 tv_array = 20 tv_alias = 21 tv_except = 22 tv_longlong = 23 tv_ulonglong = 24 tv_longdouble = 25 tv_wchar = 26 tv_wstring = 27 tv_fixed = 28 tv_value = 29 tv_value_box = 30 tv_native = 31 tv_abstract_interface = 32 tv__indirect = 0xffffffff # Create a TypeCode given a class or a repoId def typeCodeFromClassOrRepoId(t): if type(t) is types.ClassType: try: t = t._NP_RepositoryId except AttributeError: raise TypeError("Class must be a CORBA class.") if type(t) is not types.StringType: raise TypeError("Argument must be CORBA class or repository id.") d = omniORB.findType(t) if d is None: raise TypeError("Unknown CORBA type.") return createTypeCode(d) # Implementations of public ORB TypeCode creation functions def createStructTC(id, name, members): dlist = [tv_struct, None, id, name] mnames = [] for m in members: mnames.append(m.name) dlist.append(m.name) dlist.append(m.type._d) str = omniORB.createUnknownStruct(id, mnames) dlist[1] = str d = tuple(dlist) insertIndirections(d) return createTypeCode(d) def createUnionTC(id, name, discriminator_type, members): mlist = [] count = 0 defused = -1 mmap = {} for m in members: val = m.label.value() if m.label.typecode().kind() == CORBA.tk_octet and val == 0: val = -1 defused = count tup = (val, m.name, m.type._d) if defused != count: mmap[val] = tup mlist.append(tup) count = count + 1 union = omniORB.createUnknownUnion(id, defused, mlist) if defused >= 0: default = mlist[defused] else: default = None d = (tv_union, union, id, name, discriminator_type._k._v, defused, tuple(mlist), default, mmap) insertIndirections(d) return createTypeCode(d) def createEnumTC(id, name, members): mlist = [] count = 0 for m in members: mlist.append(omniORB.EnumItem(m, count)) count = count + 1 d = (tv_enum, id, name, tuple(mlist)) insertIndirections(d) return createTypeCode(d) def createAliasTC(id, name, original_type): d = (tv_alias, id, name, original_type._d) insertIndirections(d) return createTypeCode(d) def createExceptionTC(id, name, members): dlist = [tv_except, None, id, name] mnames = [] for m in members: mnames.append(m.name) dlist.append(m.name) dlist.append(m.type._d) exc = omniORB.createUnknownException(id, mnames) dlist[1] = exc d = tuple(dlist) insertIndirections(d) return createTypeCode(d) def createInterfaceTC(id, name): d = (tv_objref, id, name) insertIndirections(d) return createTypeCode(d) def createStringTC(bound): d = (tv_string, bound) insertIndirections(d) return createTypeCode(d) def createSequenceTC(bound, element_type): d = (tv_sequence, element_type._d, bound) insertIndirections(d) return createTypeCode(d) def createArrayTC(length, element_type): d = (tv_array, element_type._d, length) insertIndirections(d) return createTypeCode(d) def createRecursiveTC(id): class recursivePlaceHolder: pass recursivePlaceHolder._d = (tv__indirect, [id]) return recursivePlaceHolder() # Function to create a TypeCode object given a descriptor. Returns a # static (stub generated) TypeCode object if possible. def createTypeCode(d, parent=None): if type(d) is types.TupleType: k = d[0] else: k = d if k == tv_null: return TypeCode_empty(d) elif k == tv_void: return TypeCode_empty(d) elif k == tv_short: return TypeCode_empty(d) elif k == tv_long: return TypeCode_empty(d) elif k == tv_ushort: return TypeCode_empty(d) elif k == tv_ulong: return TypeCode_empty(d) elif k == tv_float: return TypeCode_empty(d) elif k == tv_double: return TypeCode_empty(d) elif k == tv_boolean: return TypeCode_empty(d) elif k == tv_char: return TypeCode_empty(d) elif k == tv_octet: return TypeCode_empty(d) elif k == tv_any: return TypeCode_empty(d) elif k == tv_TypeCode: return TypeCode_empty(d) elif k == tv_Principal: return TypeCode_empty(d) elif k == tv_longlong: return TypeCode_empty(d) elif k == tv_ulonglong: return TypeCode_empty(d) elif k == tv_longdouble:return TypeCode_empty(d) elif k == tv_string: return TypeCode_string(d) elif k == tv_objref: tc = omniORB.findTypeCode(d[1]) if tc is None: tc = TypeCode_objref(d) return tc elif k == tv_struct: tc = omniORB.findTypeCode(d[2]) if tc is None: tc = TypeCode_struct(d, parent) return tc elif k == tv_union: tc = omniORB.findTypeCode(d[2]) if tc is None: tc = TypeCode_union(d, parent) return tc elif k == tv_enum: tc = omniORB.findTypeCode(d[1]) if tc is None: tc = TypeCode_enum(d) return tc elif k == tv_sequence: return TypeCode_sequence(d, parent) elif k == tv_array: return TypeCode_array(d, parent) elif k == tv_alias: tc = omniORB.findTypeCode(d[1]) if tc is None: tc = TypeCode_alias(d, parent) return tc elif k == tv_except: tc = omniORB.findTypeCode(d[2]) if tc is None: tc = TypeCode_except(d, parent) return tc elif k == tv__indirect: return createTypeCode(d[1][0], parent) raise CORBA.INTERNAL() # TypeCode base interface class TypeCode_base (CORBA.TypeCode): def __init__(self): self._d = 0 self._k = CORBA.tk_null def equal(self, tc): try: if self._d == tc._d: return CORBA.TRUE else: return CORBA.FALSE except AttributeError: raise CORBA.BAD_PARAM() def equivalent(self, tc): return self.equal(tc) def get_compact_typecode(self): return self def kind(self): return self._k # Operations which are only available for some kinds: def id(self): raise CORBA.TypeCode.BadKind() def name(self): raise CORBA.TypeCode.BadKind() def member_count(self): raise CORBA.TypeCode.BadKind() def member_name(self, index): raise CORBA.TypeCode.BadKind() def member_type(self, index): raise CORBA.TypeCode.BadKind() def member_label(self, index): raise CORBA.TypeCode.BadKind() def discriminator_type(self): raise CORBA.TypeCode.BadKind() def default_index(self): raise CORBA.TypeCode.BadKind() def length(self): raise CORBA.TypeCode.BadKind() def content_type(self): raise CORBA.TypeCode.BadKind() # Things for types we don't support: def fixed_digits(self): raise CORBA.TypeCode.BadKind() def fixed_scale(self): raise CORBA.TypeCode.BadKind() def member_visibility(self, index): raise CORBA.TypeCode.BadKind() def type_modifier(self): raise CORBA.TypeCode.BadKind() def concrete_base_type(self): raise CORBA.TypeCode.BadKind() # Class for short, long, ushort, ulong, float, double, boolean, char, # octet, any, TypeCode, Principal, longlong, ulonglong, longdouble: class TypeCode_empty (TypeCode_base): def __init__(self, desc): if type(desc) is not types.IntType: raise CORBA.INTERNAL() if desc != tv_null and \ desc != tv_void and \ desc != tv_short and \ desc != tv_long and \ desc != tv_ushort and \ desc != tv_ulong and \ desc != tv_float and \ desc != tv_double and \ desc != tv_boolean and \ desc != tv_char and \ desc != tv_octet and \ desc != tv_any and \ desc != tv_TypeCode and \ desc != tv_Principal and \ desc != tv_longlong and \ desc != tv_ulonglong and \ desc != tv_longdouble: raise CORBA.INTERNAL() self._d = desc self._k = CORBA.TCKind._item(desc) # string: class TypeCode_string (TypeCode_base): def __init__(self, desc): if type(desc) is not types.TupleType or \ desc[0] != tv_string: raise CORBA.INTERNAL() self._d = desc self._k = CORBA.tk_string def length(self): return self._d[1] # objref: class TypeCode_objref (TypeCode_base): def __init__(self, desc): if type(desc) is not types.TupleType or \ desc[0] != tv_objref: raise CORBA.INTERNAL() self._d = desc self._k = CORBA.tk_objref def id(self): if self._d[1] is not None: return self._d[1] else: return "" def name(self): return self._d[2] # struct: class TypeCode_struct (TypeCode_base): def __init__(self, desc, parent): if type(desc) is not types.TupleType or \ desc[0] != tv_struct: raise CORBA.INTERNAL() self._d = desc self._k = CORBA.tk_struct self._p = parent def __del__(self): if self._p is None and removeIndirections is not None: removeIndirections(self._d) def equivalent(self, tc): return equivalentDescriptors(self._d, tc._d) def get_compact_typecode(self): return TypeCode_struct(getCompactDescriptor(self._d), None) def id(self): return self._d[2] def name(self): return self._d[3] def member_count(self): return (len(self._d) - 4) / 2 def member_name(self, index): off = index * 2 + 4 if index < 0 or off >= len(self._d): raise CORBA.TypeCode.Bounds() return self._d[off] def member_type(self, index): off = index * 2 + 5 if index < 0 or off >= len(self._d): raise CORBA.TypeCode.Bounds() if self._p is None and removeIndirections is not None: return createTypeCode(self._d[off], self) else: return createTypeCode(self._d[off], self._p) # union: class TypeCode_union (TypeCode_base): def __init__(self, desc, parent): if type(desc) is not types.TupleType or \ desc[0] != tv_union: raise CORBA.INTERNAL() self._d = desc self._k = CORBA.tk_union self._p = parent def __del__(self): if self._p is None and removeIndirections is not None: removeIndirections(self._d) def equivalent(self, tc): return equivalentDescriptors(self._d, tc._d) def get_compact_typecode(self): return TypeCode_union(getCompactDescriptor(self._d), None) def id(self): return self._d[2] def name(self): return self._d[3] def member_count(self): return len(self._d[6]) def member_name(self, index): if index < 0 or index >= len(self._d[6]): raise CORBA.TypeCode.Bounds() return self._d[6][index][1] def member_type(self, index): if index < 0 or index >= len(self._d[6]): raise CORBA.TypeCode.Bounds() if self._p is None and removeIndirections is not None: return createTypeCode(self._d[6][index][2], self) else: return createTypeCode(self._d[6][index][2], self._p) def member_label(self, index): if index < 0 or index >= len(self._d[6]): raise CORBA.TypeCode.Bounds() if index == self._d[5]: return CORBA.Any(CORBA._tc_octet, 0) return CORBA.Any(createTypeCode(self._d[4]), self._d[6][index][0]) def discriminator_type(self): return createTypeCode(self._d[4]) def default_index(self): if self._d[5] >= 0: return self._d[5] return -1 # enum: class TypeCode_enum (TypeCode_base): def __init__(self, desc): if type(desc) is not types.TupleType or \ desc[0] != tv_enum: raise CORBA.INTERNAL() self._d = desc self._k = CORBA.tk_enum def equivalent(self, tc): return equivalentDescriptors(self._d, tc._d) def get_compact_typecode(self): return TypeCode_enum(getCompactDescriptor(self._d)) def id(self): return self._d[1] def name(self): return self._d[2] def member_count(self): return len(self._d[3]) def member_name(self, index): if index < 0 or index >= len(self._d[3]): raise CORBA.TypeCode.Bounds() return self._d[3][index]._n # sequence: class TypeCode_sequence (TypeCode_base): def __init__(self, desc, parent): if type(desc) is not types.TupleType or \ desc[0] != tv_sequence: raise CORBA.INTERNAL() self._d = desc self._k = CORBA.tk_sequence self._p = parent def __del__(self): if self._p is None and removeIndirections is not None: removeIndirections(self._d) def equivalent(self, tc): return equivalentDescriptors(self._d, tc._d) def get_compact_typecode(self): return TypeCode_sequence(getCompactDescriptor(self._d), None) def length(self): return self._d[2] def content_type(self): if self._p is None and removeIndirections is not None: return createTypeCode(self._d[1], self) else: return createTypeCode(self._d[1], self._p) # array: class TypeCode_array (TypeCode_base): def __init__(self, desc, parent): if type(desc) is not types.TupleType or \ desc[0] != tv_array: raise CORBA.INTERNAL() self._d = desc self._k = CORBA.tk_array self._p = parent def __del__(self): if self._p is None and removeIndirections is not None: removeIndirections(self._d) def equivalent(self, tc): return equivalentDescriptors(self._d, tc._d) def get_compact_typecode(self): return TypeCode_sequence(getCompactDescriptor(self._d), None) def length(self): return self._d[2] def content_type(self): return createTypeCode(self._d[1]) # alias: class TypeCode_alias (TypeCode_base): def __init__(self, desc, parent): if type(desc) is not types.TupleType or \ desc[0] != tv_alias: raise CORBA.INTERNAL() self._d = desc self._k = CORBA.tk_alias self._p = parent def __del__(self): if self._p is None and removeIndirections is not None: removeIndirections(self._d) def equivalent(self, tc): return equivalentDescriptors(self._d, tc._d) def get_compact_typecode(self): return TypeCode_alias(getCompactDescriptor(self._d), None) def id(self): return self._d[1] def name(self): return self._d[2] def content_type(self): return createTypeCode(self._d[3]) # except: class TypeCode_except (TypeCode_base): def __init__(self, desc, parent): if type(desc) is not types.TupleType or \ desc[0] != tv_except: raise CORBA.INTERNAL() self._d = desc self._k = CORBA.tk_except self._p = parent def __del__(self): if self._p is None and removeIndirections is not None: removeIndirections(self._d) def equivalent(self, tc): return equivalentDescriptors(self._d, tc._d) def get_compact_typecode(self): return TypeCode_except(getCompactDescriptor(self._d), None) def id(self): return self._d[2] def name(self): return self._d[3] def member_count(self): return (len(self._d) - 4) / 2 def member_name(self, index): off = index * 2 + 4 if index < 0 or off >= len(self._d): raise CORBA.TypeCode.Bounds() return self._d[off] def member_type(self, index): off = index * 2 + 5 if index < 0 or off >= len(self._d): raise CORBA.TypeCode.Bounds() if self._p is None and removeIndirections is not None: return createTypeCode(self._d[off], self) else: return createTypeCode(self._d[off], self._p) # Functions to test descriptor equivalent def equivalentDescriptors(a, b, seen=None): if seen is None: seen = {} try: if a == b: return 1 # If they don't trivially match, they must be tuples: if type(a) is not types.TupleType or type(b) is not types.TupleType: return 0 # Follow aliases and indirections while a[0] == tv_alias or a[0] == tv__indirect: if a[0] == tv_alias: a = a[3] else: a = a[1][0] while b[0] == tv_alias or b[0] == tv__indirect: if b[0] == tv_alias: b = b[3] else: b = b[1][0] if seen.has_key((id(a),id(b))): return 1 seen[id(a),id(b)] = None # Must be same kind if a[0] != b[0]: return 0 if a[0] == tv_struct: # id if a[2] != "" and b[2] != "": if a[2] == b[2]: return 1 else: return 0 # members: if len(a) != len(b): return 0 for i in range(4, len(a), 2): # Member type if not equivalentDescriptors(a[i+1], b[i+1], seen): return 0 return 1 elif a[0] == tv_union: # id if a[2] != "" and b[2] != "": if a[2] == b[2]: return 1 else: return 0 # discriminant type if not equivalentDescriptors(a[4], b[4], seen): return 0 # default index if a[5] != b[5]: return 0 # Members if len(a[6]) != len(b[6]): return 0 for i in range(len(a[6])): # Member label if a[6][i][0] != b[6][i][0]: return 0 # Member descriptor if not equivalentDescriptors(a[6][i][2], b[6][i][2], seen): return 0 return 1 elif a[0] == tv_enum: # id if a[1] != "" and b[1] != "": if a[1] == b[1]: return 1 else: return 0 # Members if len(a[3]) != len(b[3]): return 0 return 1 elif a[0] == tv_sequence: # Bound if a[2] != b[2]: return 0 # Type return equivalentDescriptors(a[1], b[1], seen) elif a[0] == tv_array: # Length if a[2] != b[2]: return 0 # Type return equivalentDescriptors(a[1], b[1], seen) elif a[0] == tv_except: # id if a[2] != "" and b[2] != "": if a[2] == b[2]: return 1 else: return 0 # members: if len(a) != len(b): return 0 for i in range(4, len(self._d), 2): # Member type if not equivalentDescriptors(a[i+1], b[i+1], seen): return 0 return 1 return 0 except AttributeError: raise CORBA.BAD_PARAM() # Functions to compact descriptors: def getCompactDescriptor(d): seen = {} ind = [] r = r_getCompactDescriptor(d, seen, ind) # Fix up indirections: for i in ind: try: i[0] = seen[id(i[0])] except KeyError: raise CORBA.BAD_TYPECODE() return r def r_getCompactDescriptor(d, seen, ind): if type(d) is types.TupleType: k = d[0] else: k = d if k == tv_short: r = d elif k == tv_long: r = d elif k == tv_ushort: r = d elif k == tv_ulong: r = d elif k == tv_float: r = d elif k == tv_double: r = d elif k == tv_boolean: r = d elif k == tv_char: r = d elif k == tv_octet: r = d elif k == tv_any: r = d elif k == tv_TypeCode: r = d elif k == tv_Principal: r = d elif k == tv_string: r = d elif k == tv_objref: r = d elif k == tv_longlong: r = d elif k == tv_ulonglong: r = d elif k == tv_longdouble:r = d elif k == tv_struct: c = list(d) c[3] = "" for i in range(4, len(c), 2): c[i] = "" c[i+1] = r_getCompactDescriptor(d[i+1], seen, ind) r = tuple(c) elif k == tv_union: c = list(d) c[3] = "" c[4] = r_getCompactDescriptor(d[4], seen, ind) m = [] for u in d[6]: m.append((u[0], "", r_getCompactDescriptor(u[2], seen, ind))) c[6] = tuple(m) if d[7] is not None: c[7] = (d[7][0], "", r_getCompactDescriptor(d[7][2], seen, ind)) r = tuple(c) elif k == tv_enum: m = [] for e in d[3]: m.append(omniORB.AnonymousEnumItem(e._v)) r = (k, d[1], "", tuple(m)) elif k == tv_sequence: r = (k, r_getCompactDescriptor(d[1], seen, ind), d[2]) elif k == tv_array: r = (k, r_getCompactDescriptor(d[1], seen, ind), d[2]) elif k == tv_alias: r = (k, d[1], "", r_getCompactDescriptor(d[3], seen, ind)) elif k == tv_except: c = list(d) c[3] = "" for i in range(4, len(c), 2): c[i] = "" c[i+1] = r_getCompactDescriptor(d[i+1], seen, ind) r = tuple(c) elif k == tv__indirect: l = [d[1][0]] ind.append(l) r = (k, l) else: raise CORBA.INTERNAL() seen[id(d)] = r return r # Function to remove indirections from a descriptor, so it can be # collected by Python's reference counting garbage collector: def removeIndirections(desc): if type(desc) is not types.TupleType: return k = desc[0] if k == tv_struct: for i in range(5, len(desc), 2): removeIndirections(desc[i]) elif k == tv_union: for t in desc[6]: removeIndirections(t[2]) if desc[7] is not None: removeIndirections(desc[7][2]) elif k == tv_sequence: removeIndirections(desc[1]) elif k == tv_array: removeIndirections(desc[1]) elif k == tv_alias: removeIndirections(desc[3]) elif k == tv_except: for i in range(5, len(desc), 2): removeIndirections(desc[i]) elif k == tv__indirect: del(desc[1][0]) # Function to insert indirections into a descriptor, replacing repoIds # with references def insertIndirections(d): seen = {} ind = [] r_insertIndirections(d, seen, ind) # Fix up indirections: for i in ind: try: i[0] = seen[i[0]] except KeyError: pass def r_insertIndirections(d, seen, ind): if type(d) is not types.TupleType: return k = d[0] if k == tv_struct: if not seen.has_key(d[2]): seen[d[2]] = d for i in range(4, len(d), 2): r_insertIndirections(d[i+1], seen, ind) elif k == tv_union: if not seen.has_key(d[2]): seen[d[2]] = d for u in d[6]: r_insertIndirections(u[2], seen, ind) elif k == tv_sequence: r_insertIndirections(d[1], seen, ind) elif k == tv_array: r_insertIndirections(d[1], seen, ind) elif k == tv_alias: if not seen.has_key(d[1]): seen[d[1]] = d r_insertIndirections(d[3], seen, ind) elif k == tv_except: if not seen.has_key(d[2]): seen[d[2]] = d for i in range(4, len(d), 2): r_insertIndirections(d[i+1], seen, ind) elif k == tv__indirect: if type(d[1][0]) == types.StringType: if d[1] not in ind: ind.append(d[1])