examples/AutoTest/CorbaNaming.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# -*- Python -*-
 
 
 
 
import omniORB.CORBA as CORBA
import CosNaming
import string
 
 
class CorbaNaming:
  """
  """
 
 
 
  
  def __init__(self, orb, name_server=None):
    self._orb = orb
    self._nameServer = ""
    self._rootContext = CosNaming.NamingContext._nil
    self._blLength = 100
 
    if name_server:
      self._nameServer = "corbaloc::" + name_server + "/NameService"
      try:
        obj = orb.string_to_object(self._nameServer)
        self._rootContext = obj._narrow(CosNaming.NamingContext)
        if CORBA.is_nil(self._rootContext):
          print("CorbaNaming: Failed to narrow the root naming context.")
 
      except CORBA.ORB.InvalidName:
        print("Service required is invalid [does not exist].")
 
    return
  
 
  
  def __del__(self):
    return
 
 
  
  def init(self, name_server):
    self._nameServer = "corbaloc::" + name_server + "/NameService"
    obj = self._orb.string_to_object(self._nameServer)
    self._rootContext = obj._narrow(CosNaming.NamingContext)
    if CORBA.is_nil(self._rootContext):
      raise MemoryError
 
    return
 
 
  
  def bind(self, name_list, obj, force=None):
    if force is None :
      force = True
 
    try:
      self._rootContext.bind(name_list, obj)
    except CosNaming.NamingContext.NotFound:
      if force:
        self.bindRecursive(self._rootContext, name_list, obj)
      else:
        raise
    except CosNaming.NamingContext.CannotProceed as err:
      if force:
        self.bindRecursive(err.cxt, err.rest_of_name, obj)
      else:
        raise
    except CosNaming.NamingContext.AlreadyBound:
      self._rootContext.rebind(name_list, obj)
 
 
  
  def bindByString(self, string_name, obj, force=True):
    self.bind(self.toName(string_name), obj, force)
 
 
  
  def bindRecursive(self, context, name_list, obj):
    length = len(name_list)
    cxt = context
    for i in range(length):
      if i == length -1:
        try:
          cxt.bind(self.subName(name_list, i, i), obj)
        except CosNaming.NamingContext.AlreadyBound:
          cxt.rebind(self.subName(name_list, i, i), obj)
        return
      else:
        if self.objIsNamingContext(cxt):
          cxt = self.bindOrResolveContext(cxt,self.subName(name_list, i, i))
        else:
          raise CosNaming.NamingContext.CannotProceed(cxt, self.subName(name_list, i))
    return
 
 
  
  def rebind(self, name_list, obj, force=True):
    if force is None:
      force = True
      
    try:
      self._rootContext.rebind(name_list, obj)
 
    except CosNaming.NamingContext.NotFound:
      if force:
        self.rebindRecursive(self._rootContext, name_list, obj)
      else:
        raise
 
    except CosNaming.NamingContext.CannotProceed as err:
      if force:
        self.rebindRecursive(err.cxt, err,rest_of_name, obj)
      else:
        raise
      
    return
 
 
  
  def rebindByString(self, string_name, obj, force=True):
    self.rebind(self.toName(string_name), obj, force)
 
    return
 
 
  
  def rebindRecursive(self, context, name_list, obj):
    length = len(name_list)
    for i in range(length):
      if i == length - 1:
        context.rebind(self.subName(name_list, i, i), obj)
        return
      else:
        if self.objIsNamingContext(context):
          try:
            context = context.bind_new_context(self.subName(name_list, i, i))
          except CosNaming.NamingContext.AlreadyBound:
            obj_ = context.resolve(self.subName(name_list, i, i))
            context = obj_._narrow(CosNaming.NamingContext)
        else:
          raise CosNaming.NamingContext.CannotProceed(context, self.subName(name_list, i))
    return
 
 
  
  def bindContext(self, name, name_cxt, force=True):
    if isinstance(name, str):
      self.bind(self.toName(name), name_cxt, force)
    else:
      self.bind(name, name_cxt, force)
    return
 
 
  
  def bindContextRecursive(self, context, name_list, name_cxt):
    self.bindRecursive(context, name_list, name_cxt)
    return
 
 
  
  def rebindContext(self, name, name_cxt, force=True):
    if isinstance(name, str):
      self.rebind(self.toName(name), name_cxt, force)
    else:
      self.rebind(name, name_cxt, force)
    return
 
 
  
  def rebindContextRecursive(self, context, name_list, name_cxt):
    self.rebindRecursive(context, name_list, name_cxt)
    return
 
 
  
  def resolve(self, name):
    if isinstance(name, str):
      name_ = self.toName(name)
    else:
      name_ = name
      
    try:
      obj = self._rootContext.resolve(name_)
      return obj
    except CosNaming.NamingContext.NotFound as ex:
      return None
 
 
  
  def unbind(self, name):
    if isinstance(name, str):
      name_ = self.toName(name)
    else:
      name_ = name
 
    self._rootContext.unbind(name_)
    return
 
 
  
  def newContext(self):
    return self._rootContext.new_context()
 
 
  
  def bindNewContext(self, name, force=True):
    if force is None:
      force = True
      
    if isinstance(name, str):
      name_ = self.toName(name)
    else:
      name_ = name
 
    try:
      return self._rootContext.bind_new_context(name_)
    except CosNaming.NamingContext.NotFound:
      if force:
        self.bindRecursive(self._rootContext, name_, self.newContext())
      else:
        raise
    except CosNaming.NamingContext.CannotProceed as err:
      if force:
        self.bindRecursive(err.cxt, err.rest_of_name, self.newContext())
      else:
        raise
    return None
 
 
  
  def destroy(self, context):
    context.destroy()
 
 
  
  def destroyRecursive(self, context):
    cont = True
    bl = []
    bi = 0
    bl, bi = context.list(self._blLength)
    while cont:
      for i in range(len(bl)):
        if bl[i].binding_type == CosNaming.ncontext:
          obj = context.resolve(bl[i].binding_name)
          next_context = obj._narrow(CosNaming.NamingContext)
 
          self.destroyRecursive(next_context)
          context.unbind(bl[i].binding_name)
          next_context.destroy()
        elif bl[i].binding_type == CosNaming.nobject:
          context.unbind(bl[i].binding_name)
        else:
          assert(0)
      if CORBA.is_nil(bi):
        cont = False
      else:
        bi.next_n(self._blLength, bl)
 
    if not (CORBA.is_nil(bi)):
      bi.destroy()
    return
 
 
  
  def clearAll(self):
    self.destroyRecursive(self._rootContext)
    return
 
 
  
  def list(self, name_cxt, how_many, rbl, rbi):
    bl, bi = name_cxt.list(how_many)
 
    for i in bl:
      rbl.append(bl)
 
    rbi.append(bi)
  
 
  #============================================================
  # interface of NamingContext
  #============================================================
 
  
  def toString(self, name_list):
    if len(name_list) == 0:
      raise CosNaming.NamingContext.InvalidName
 
    slen = self.getNameLength(name_list)
    string_name = [""]
    self.nameToString(name_list, string_name, slen)
 
    return string_name
 
 
  
  def toName(self, sname):
    if not sname:
      raise CosNaming.NamingContext.InvalidName
 
    string_name = sname
    name_comps = []
 
    nc_length = 0
    nc_length = self.split(string_name, "/", name_comps)
    if not (nc_length > 0):
      raise CosNaming.NamingContext.InvalidName
 
    name_list = [CosNaming.NameComponent("","") for i in range(nc_length)]
 
    for i in range(nc_length):
      pos = string.rfind(name_comps[i][0:],".")
      if pos == -1:
        name_list[i].id   = name_comps[i]
        name_list[i].kind = ""
      else:
        name_list[i].id   = name_comps[i][0:pos]
        name_list[i].kind = name_comps[i][(pos+1):]
 
    return name_list
 
 
  
  def toUrl(self, addr, string_name):
    return self._rootContext.to_url(addr, string_name)
 
 
  
  def resolveStr(self, string_name):
    return self.resolve(self.toName(string_name))
 
 
  #============================================================
  # Find functions
  #============================================================
 
  
  def bindOrResolve(self, context, name_list, obj):
    try:
      context.bind_context(name_list, obj)
      return obj
    except CosNaming.NamingContext.AlreadyBound:
      obj = context.resolve(name_list)
      return obj
    return CORBA.Object._nil
 
 
  
  def bindOrResolveContext(self, context, name_list, new_context=None):
    if new_context is None:
      new_cxt = self.newContext()
    else:
      new_cxt = new_context
 
    obj = self.bindOrResolve(context, name_list, new_cxt)
    return obj._narrow(CosNaming.NamingContext)
 
 
  
  def getNameServer(self):
    return self._nameServer
 
 
  
  def getRootContext(self):
    return self._rootContext
 
 
  
  def objIsNamingContext(self, obj):
    nc = obj._narrow(CosNaming.NamingContext)
    if CORBA.is_nil(nc):
      return False
    else:
      return True
 
 
  
  def nameIsNamingContext(self, name_list):
    return self.objIsNamingContext(self.resolve(name_list))
 
 
  
  def subName(self, name_list, begin, end = None):
    if end is None or end < 0:
      end = len(name_list) - 1
 
    sub_len = end - (begin -1)
    objId = ""
    kind  = ""
    
    sub_name = []
    for i in range(sub_len):
      sub_name.append(name_list[begin + i])
 
    return sub_name
 
 
  
  def nameToString(self, name_list, string_name, slen):
    for i in range(len(name_list)):
      for id_ in name_list[i].id:
        if id_ == "/" or id_ == "." or id_ == "\\":
          string_name[0] += "\\"
        string_name[0] += id_
 
      if name_list[i].id == "" or name_list[i].kind != "":
        string_name[0] += "."
 
      for kind_ in name_list[i].kind:
        if kind_ == "/" or kind_ == "." or kind_ == "\\":
          string_name[0] += "\\"
        string_name[0] += kind_
 
      string_name[0] += "/"
 
 
  
  def getNameLength(self, name_list):
    slen = 0
 
    for i in range(len(name_list)):
      for id_ in name_list[i].id:
        if id_ == "/" or id_ == "." or id_ == "\\":
          slen += 1
        slen += 1
      if name_list[i].id == "" or name_list[i].kind == "":
        slen += 1
 
      for kind_ in name_list[i].kind:
        if kind_ == "/" or kind_ == "." or kind_ == "\\":
          slen += 1
        slen += 1
 
      slen += 1
 
    return slen
 
 
  
  def split(self, input, delimiter, results):
    delim_size = len(delimiter)
    found_pos = begin_pos = pre_pos = substr_size = 0
 
    if input[0:delim_size] == delimiter:
      begin_pos = pre_pos = delim_size
 
    while 1:
      found_pos = string.find(input[begin_pos:],delimiter)
      
      if found_pos == -1:
        results.append(input[pre_pos:])
        break
 
      if found_pos > 0 and input[found_pos - 1] == "\\":
        begin_pos += found_pos + delim_size
      else:
        substr_size = found_pos + (begin_pos - pre_pos)
        if substr_size > 0:
          results.append(input[pre_pos:(pre_pos+substr_size)])
        begin_pos += found_pos + delim_size
        pre_pos   = begin_pos
 
    return len(results)