# See the License for the specific language governing permissions and
# limitations under the License.
-from threading import Event
from Queue import Queue
+from threading import Thread, Lock, Event
-from ncclient.capabilities import Capabilities, CAPABILITIES
-from ncclient.glue import Subject
+from ncclient import xml_
+from ncclient.capabilities import Capabilities
-from hello import HelloHandler
+from errors import TransportError
import logging
logger = logging.getLogger('ncclient.transport.session')
-class Session(Subject):
-
- "TODO: docstring"
-
- def __init__(self):
- "Subclass constructor should call this"
- Subject.__init__(self)
+class Session(Thread):
+
+ "Base class for use by transport protocol implementations."
+
+ def __init__(self, capabilities):
+ Thread.__init__(self)
+ self.setDaemon(True)
+ self._listeners = set() # 3.0's weakset would be ideal
+ self._lock = Lock()
self.setName('session')
self._q = Queue()
- self._client_capabilities = CAPABILITIES
+ self._client_capabilities = capabilities
self._server_capabilities = None # yet
self._id = None # session-id
self._connected = False # to be set/cleared by subclass implementation
logger.debug('%r created: client_capabilities=%r' %
(self, self._client_capabilities))
-
+
+ def _dispatch_message(self, raw):
+ try:
+ root = xml_.parse_root(raw)
+ except Exception as e:
+ logger.error('error parsing dispatch message: %s' % e)
+ return
+ with self._lock:
+ listeners = list(self._listeners)
+ for l in listeners:
+ logger.debug('dispatching message to %r' % l)
+ try:
+ l.callback(root, raw)
+ except Exception as e:
+ logger.warning('[error] %r' % e)
+
+ def _dispatch_error(self, err):
+ with self._lock:
+ listeners = list(self._listeners)
+ for l in listeners:
+ logger.debug('dispatching error to %r' % l)
+ try:
+ l.errback(err)
+ except Exception as e:
+ logger.warning('error dispatching to %r: %r' % (l, e))
+
def _post_connect(self):
"Greeting stuff"
init_event = Event()
# callbacks
def ok_cb(id, capabilities):
self._id = id
- self._server_capabilities = Capabilities(capabilities)
+ self._server_capabilities = capabilities
init_event.set()
def err_cb(err):
error[0] = err
self.remove_listener(listener)
if error[0]:
raise error[0]
+ #if ':base:1.0' not in self.server_capabilities:
+ # raise MissingCapabilityError(':base:1.0')
logger.info('initialized: session-id=%s | server_capabilities=%s' %
- (self._id, self._server_capabilities))
-
- def connect(self, *args, **kwds):
- "Subclass implements"
+ (self._id, self._server_capabilities))
+
+ def add_listener(self, listener):
+ """Register a listener that will be notified of incoming messages and
+ errors.
+
+ :type listener: :class:`SessionListener`
+ """
+ logger.debug('installing listener %r' % listener)
+ if not isinstance(listener, SessionListener):
+ raise SessionError("Listener must be a SessionListener type")
+ with self._lock:
+ self._listeners.add(listener)
+
+ def remove_listener(self, listener):
+ """Unregister some listener; ignore if the listener was never
+ registered.
+
+ :type listener: :class:`SessionListener`
+ """
+ logger.debug('discarding listener %r' % listener)
+ with self._lock:
+ self._listeners.discard(listener)
+
+ def get_listener_instance(self, cls):
+ """If a listener of the specified type is registered, returns the
+ instance.
+
+ :type cls: :class:`SessionListener`
+ """
+ with self._lock:
+ for listener in self._listeners:
+ if isinstance(listener, cls):
+ return listener
+
+ def connect(self, *args, **kwds): # subclass implements
raise NotImplementedError
- def run(self):
- "Subclass implements"
+ def run(self): # subclass implements
raise NotImplementedError
-
+
def send(self, message):
- "TODO: docstring"
+ """Send the supplied *message* to NETCONF server.
+
+ :arg message: an XML document
+
+ :type message: `string`
+ """
+ if not self.connected:
+ raise TransportError('Not connected to NETCONF server')
logger.debug('queueing %s' % message)
self._q.put(message)
-
+
### Properties
-
+
+ @property
+ def connected(self):
+ "Connection status of the session."
+ return self._connected
+
@property
def client_capabilities(self):
+ "Client's :class:`Capabilities`"
return self._client_capabilities
-
+
@property
def server_capabilities(self):
+ "Server's :class:`Capabilities`"
return self._server_capabilities
-
- @property
- def connected(self):
- return self._connected
-
+
@property
def id(self):
+ """A `string` representing the `session-id`. If the session has not
+ been initialized it will be :const:`None`"""
return self._id
-
+
@property
def can_pipeline(self):
+ "Whether this session supports pipelining"
return True
+
+
+class SessionListener(object):
+
+ """Base class for :class:`Session` listeners, which are notified when a new
+ NETCONF message is received or an error occurs.
+
+ .. note::
+ Avoid time-intensive tasks in a callback's context.
+ """
+
+ def callback(self, root, raw):
+ """Called when a new XML document is received. The `root` argument
+ allows the callback to determine whether it wants to further process the
+ document.
+
+ :arg root: tuple of `(tag, attributes)` where `tag` is the qualified name of the root element and `attributes` is a dictionary of its attributes (also qualified names)
+ :type root: `tuple`
+
+ :arg raw: XML document
+ :type raw: `string`
+ """
+ raise NotImplementedError
+
+ def errback(self, ex):
+ """Called when an error occurs.
+
+ :type ex: :exc:`Exception`
+ """
+ raise NotImplementedError
+
+
+class HelloHandler(SessionListener):
+
+ def __init__(self, init_cb, error_cb):
+ self._init_cb = init_cb
+ self._error_cb = error_cb
+
+ def callback(self, root, raw):
+ if xml_.unqualify(root[0]) == 'hello':
+ try:
+ id, capabilities = HelloHandler.parse(raw)
+ except Exception as e:
+ self._error_cb(e)
+ else:
+ self._init_cb(id, capabilities)
+
+ def errback(self, err):
+ self._error_cb(err)
+
+ @staticmethod
+ def build(capabilities):
+ "Given a list of capability URI's returns <hello> message XML string"
+ spec = {
+ 'tag': 'hello',
+ 'attrib': {'xmlns': xml_.BASE_NS},
+ 'subtree': [{
+ 'tag': 'capabilities',
+ 'subtree': # this is fun :-)
+ [{'tag': 'capability', 'text': uri} for uri in capabilities]
+ }]
+ }
+ return xml_.dtree2xml(spec)
+
+ @staticmethod
+ def parse(raw):
+ "Returns tuple of (session-id (str), capabilities (Capabilities)"
+ sid, capabilities = 0, []
+ root = xml_.xml2ele(raw)
+ for child in root.getchildren():
+ tag = xml_.unqualify(child.tag)
+ if tag == 'session-id':
+ sid = child.text
+ elif tag == 'capabilities':
+ for cap in child.getchildren():
+ if xml_.unqualify(cap.tag) == 'capability':
+ capabilities.append(cap.text)
+ return sid, Capabilities(capabilities)
projects / ncclient / blobdiff