summaryrefslogtreecommitdiff
path: root/qpid/extras/dispatch/src/py
diff options
context:
space:
mode:
authorTed Ross <tross@apache.org>2013-08-01 20:19:24 +0000
committerTed Ross <tross@apache.org>2013-08-01 20:19:24 +0000
commit437b93efc7f559a94f05e552527795c1df68516c (patch)
tree549413e09e043b825f33991f23ec99538bcaf73e /qpid/extras/dispatch/src/py
parent7919bb3eb92d85f9dcb3d6e3ada2370f3b419bb2 (diff)
downloadqpid-python-437b93efc7f559a94f05e552527795c1df68516c.tar.gz
QPID-4967 - Updates to the router
- The router module now tracks other router nodes (neighbors and non-neighbors) - Tracked nodes are communicated to the router_node.c fast-path git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1509415 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/extras/dispatch/src/py')
-rw-r--r--qpid/extras/dispatch/src/py/router/link.py3
-rw-r--r--qpid/extras/dispatch/src/py/router/neighbor.py2
-rw-r--r--qpid/extras/dispatch/src/py/router/node.py103
-rw-r--r--qpid/extras/dispatch/src/py/router/router_engine.py25
4 files changed, 133 insertions, 0 deletions
diff --git a/qpid/extras/dispatch/src/py/router/link.py b/qpid/extras/dispatch/src/py/router/link.py
index 1e06d161f6..ce97794b8c 100644
--- a/qpid/extras/dispatch/src/py/router/link.py
+++ b/qpid/extras/dispatch/src/py/router/link.py
@@ -87,6 +87,7 @@ class LinkStateEngine(object):
self.collection[msg.id] = ls
self.collection_changed = True
ls.last_seen = now
+ self.container.new_node(msg.id)
self.container.log(LOG_INFO, "Learned link-state from new router: %s" % msg.id)
# Schedule LSRs for any routers referenced in this LS that we don't know about
for _id in msg.ls.peers:
@@ -108,6 +109,7 @@ class LinkStateEngine(object):
self.collection_changed = True
self._send_ra()
+
def set_mobile_sequence(self, seq):
self.mobile_seq = seq
@@ -124,6 +126,7 @@ class LinkStateEngine(object):
for key in to_delete:
ls = self.collection.pop(key)
self.collection_changed = True
+ self.container.lost_node(key)
self.container.log(LOG_INFO, "Expired link-state from router: %s" % key)
diff --git a/qpid/extras/dispatch/src/py/router/neighbor.py b/qpid/extras/dispatch/src/py/router/neighbor.py
index 55c6bab62f..8d0dfceecf 100644
--- a/qpid/extras/dispatch/src/py/router/neighbor.py
+++ b/qpid/extras/dispatch/src/py/router/neighbor.py
@@ -64,6 +64,7 @@ class NeighborEngine(object):
if msg.is_seen(self.id):
if self.link_state.add_peer(msg.id):
self.link_state_changed = True
+ self.container.new_neighbor(msg.id)
self.container.log(LOG_INFO, "New neighbor established: %s" % msg.id)
##
## TODO - Use this function to detect area boundaries
@@ -78,6 +79,7 @@ class NeighborEngine(object):
self.hellos.pop(key)
if self.link_state.del_peer(key):
self.link_state_changed = True
+ self.container.lost_neighbor(key)
self.container.log(LOG_INFO, "Neighbor lost: %s" % key)
diff --git a/qpid/extras/dispatch/src/py/router/node.py b/qpid/extras/dispatch/src/py/router/node.py
new file mode 100644
index 0000000000..482a83dfe8
--- /dev/null
+++ b/qpid/extras/dispatch/src/py/router/node.py
@@ -0,0 +1,103 @@
+#
+# 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.
+#
+
+try:
+ from dispatch import *
+except ImportError:
+ from stubs import *
+
+
+class NodeTracker(object):
+ """
+ This module is responsible for tracking the set of router nodes that are known to this
+ router. It tracks whether they are neighbor or remote and whether they are reachable.
+ """
+ def __init__(self, container):
+ self.container = container
+ self.id = self.container.id
+ self.area = self.container.area
+ self.nodes = {} # id => RemoteNode
+
+
+ def tick(self, now):
+ pass
+
+
+ def new_neighbor(self, node_id):
+ if node_id not in self.nodes:
+ self.nodes[node_id] = RemoteNode(node_id)
+ self.nodes[node_id].set_neighbor()
+ self._notify(self.nodes[node_id])
+
+
+ def lost_neighbor(self, node_id):
+ node = self.nodes[node_id]
+ node.clear_neighbor()
+ self._notify(node)
+ if node.to_delete():
+ self.nodes.pop(node_id)
+
+
+ def new_node(self, node_id):
+ if node_id not in self.nodes:
+ self.nodes[node_id] = RemoteNode(node_id)
+ self.nodes[node_id].set_remote()
+ self._notify(self.nodes[node_id])
+
+
+ def lost_node(self, node_id):
+ node = self.nodes[node_id]
+ node.clear_remote()
+ self._notify(node)
+ if node.to_delete():
+ self.nodes.pop(node_id)
+
+
+ def _notify(self, node):
+ if node.to_delete():
+ self.container.adapter.node_updated("R%s" % node.id, 0, 0)
+ else:
+ is_neighbor = 0
+ if node.neighbor:
+ is_neighbor = 1
+ self.container.adapter.node_updated("R%s" % node.id, 1, is_neighbor)
+
+
+class RemoteNode(object):
+
+ def __init__(self, node_id):
+ self.id = node_id
+ self.neighbor = None
+ self.remote = None
+
+ def set_neighbor(self):
+ self.neighbor = True
+
+ def set_remote(self):
+ self.remote = True
+
+ def clear_neighbor(self):
+ self.neighbor = None
+
+ def clear_remote(self):
+ self.remote = None
+
+ def to_delete(self):
+ return self.neighbor or self.remote
+
diff --git a/qpid/extras/dispatch/src/py/router/router_engine.py b/qpid/extras/dispatch/src/py/router/router_engine.py
index 065204ad62..675d9c32e5 100644
--- a/qpid/extras/dispatch/src/py/router/router_engine.py
+++ b/qpid/extras/dispatch/src/py/router/router_engine.py
@@ -29,6 +29,7 @@ from mobile import MobileAddressEngine
from routing import RoutingTableEngine
from binding import BindingEngine
from adapter import AdapterEngine
+from node import NodeTracker
##
## Import the Dispatch adapters from the environment. If they are not found
@@ -79,6 +80,7 @@ class RouterEngine:
self.routing_table_engine = RoutingTableEngine(self)
self.binding_engine = BindingEngine(self)
self.adapter_engine = AdapterEngine(self)
+ self.node_tracker = NodeTracker(self)
@@ -125,6 +127,7 @@ class RouterEngine:
self.routing_table_engine.tick(now)
self.binding_engine.tick(now)
self.adapter_engine.tick(now)
+ self.node_tracker.tick(now)
except Exception, e:
self.log(LOG_ERROR, "Exception in timer processing: exception=%r" % e)
@@ -222,6 +225,12 @@ class RouterEngine:
self.log(LOG_TRACE, "SENT: %r dest=%s" % (msg, dest))
+ def node_updated(self, addr, reachable, neighbor):
+ """
+ """
+ self.router_adapter(addr, reachable, neighbor)
+
+
##========================================================================================
## Interconnect between the Sub-Modules
##========================================================================================
@@ -253,3 +262,19 @@ class RouterEngine:
self.log(LOG_DEBUG, "Event: remote_routes_changed: class=%s routes=%r" % (key_class, routes))
self.adapter_engine.remote_routes_changed(key_class, routes)
+ def new_neighbor(self, rid):
+ self.log(LOG_DEBUG, "Event: new_neighbor: id=%s" % rid)
+ self.node_tracker.new_neighbor(rid)
+
+ def lost_neighbor(self, rid):
+ self.log(LOG_DEBUG, "Event: lost_neighbor: id=%s" % rid)
+ self.node_tracker.lost_neighbor(rid)
+
+ def new_node(self, rid):
+ self.log(LOG_DEBUG, "Event: new_node: id=%s" % rid)
+ self.node_tracker.new_node(rid)
+
+ def lost_node(self, rid):
+ self.log(LOG_DEBUG, "Event: lost_node: id=%s" % rid)
+ self.node_tracker.lost_node(rid)
+