Commit 921c7ba8 authored by Grégory Wisniewski's avatar Grégory Wisniewski

Use a decorator to check if the master connection is well established. Raise

NotReady exception to let the handler dispatch method send the answer.
NotReady can now receive a custom message.


git-svn-id: https://svn.erp5.org/repos/neo/branches/prototype3@1208 71dcc9de-d417-0410-9af5-da40c76e7ee4
parent 7cda9241
...@@ -23,24 +23,25 @@ from neo import protocol ...@@ -23,24 +23,25 @@ from neo import protocol
from neo.exception import PrimaryFailure from neo.exception import PrimaryFailure
from neo.util import dump from neo.util import dump
def master_connection_required(handler):
""" Check if the master connection is established """
def decorator(self, *args, **kwargs):
if self.app.master_conn is None:
raise protocol.NotReadyError('Not connected to a primary master.')
handler(self, *args, **kwargs)
return decorator
class AdminEventHandler(EventHandler): class AdminEventHandler(EventHandler):
"""This class deals with events for administrating cluster.""" """This class deals with events for administrating cluster."""
def __notConnected(self, conn, packet): @master_connection_required
conn.answer(protocol.notReady('Not connected to a primary master.'),
packet.getId())
def handleAskPartitionList(self, conn, packet, min_offset, max_offset, uuid): def handleAskPartitionList(self, conn, packet, min_offset, max_offset, uuid):
logging.info("ask partition list from %s to %s for %s" %(min_offset, max_offset, dump(uuid))) logging.info("ask partition list from %s to %s for %s" %(min_offset, max_offset, dump(uuid)))
app = self.app app = self.app
# check we have one pt otherwise ask it to PMN # check we have one pt otherwise ask it to PMN
if app.pt is None: if app.pt is None:
master_conn = self.app.master_conn
if master_conn is None:
self.__notConnected(conn, packet)
else:
p = protocol.askPartitionTable([]) p = protocol.askPartitionTable([])
msg_id = master_conn.ask(p) msg_id = self.app.master_conn.ask(p)
app.dispatcher.register(msg_id, conn, app.dispatcher.register(msg_id, conn,
{'min_offset' : min_offset, {'min_offset' : min_offset,
'max_offset' : max_offset, 'max_offset' : max_offset,
...@@ -66,6 +67,7 @@ class AdminEventHandler(EventHandler): ...@@ -66,6 +67,7 @@ class AdminEventHandler(EventHandler):
p = protocol.answerNodeList(node_information_list) p = protocol.answerNodeList(node_information_list)
conn.answer(p, packet.getId()) conn.answer(p, packet.getId())
@master_connection_required
def handleSetNodeState(self, conn, packet, uuid, state, modify_partition_table): def handleSetNodeState(self, conn, packet, uuid, state, modify_partition_table):
logging.info("set node state for %s-%s" %(dump(uuid), state)) logging.info("set node state for %s-%s" %(dump(uuid), state))
node = self.app.nm.getNodeByUUID(uuid) node = self.app.nm.getNodeByUUID(uuid)
...@@ -79,53 +81,38 @@ class AdminEventHandler(EventHandler): ...@@ -79,53 +81,38 @@ class AdminEventHandler(EventHandler):
conn.answer(p, packet.getId()) conn.answer(p, packet.getId())
return return
# forward to primary master node # forward to primary master node
master_conn = self.app.master_conn
if master_conn is None:
self.__notConnected(conn, packet)
else:
p = protocol.setNodeState(uuid, state, modify_partition_table) p = protocol.setNodeState(uuid, state, modify_partition_table)
msg_id = master_conn.ask(p) msg_id = self.app.master_conn.ask(p)
self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()}) self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()})
@master_connection_required
def handleSetClusterState(self, conn, packet, state): def handleSetClusterState(self, conn, packet, state):
# forward to primary # forward to primary
master_conn = self.app.master_conn
if master_conn is None:
self.__notConnected(conn, packet)
else:
p = protocol.setClusterState(state) p = protocol.setClusterState(state)
msg_id = master_conn.ask(p) msg_id = self.app.master_conn.ask(p)
self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()}) self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()})
@master_connection_required
def handleAddPendingNodes(self, conn, packet, uuid_list): def handleAddPendingNodes(self, conn, packet, uuid_list):
logging.info('Add nodes %s' % [dump(uuid) for uuid in uuid_list]) logging.info('Add nodes %s' % [dump(uuid) for uuid in uuid_list])
# forward the request to primary # forward the request to primary
master_conn = self.app.master_conn msg_id = self.app.master_conn.ask(protocol.addPendingNodes(uuid_list))
if master_conn is None:
self.__notConnected(conn, packet)
else:
msg_id = master_conn.ask(protocol.addPendingNodes(uuid_list))
self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()}) self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()})
@master_connection_required
def handleAskClusterState(self, conn, packet): def handleAskClusterState(self, conn, packet):
if self.app.cluster_state is None: if self.app.cluster_state is None:
# required it from PMN first # required it from PMN first
master_conn = self.app.master_conn msg_id = self.app.master_conn.ask(protocol.askClusterState())
if master_conn is None:
self.__notConnected(conn, packet)
else:
msg_id = master_conn.ask(protocol.askClusterState())
self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()}) self.app.dispatcher.register(msg_id, conn, {'msg_id' : packet.getId()})
return else:
conn.answer(protocol.answerClusterState(self.app.cluster_state), packet.getId()) conn.answer(protocol.answerClusterState(self.app.cluster_state),
packet.getId())
@master_connection_required
def handleAskPrimaryMaster(self, conn, packet): def handleAskPrimaryMaster(self, conn, packet):
master_node = self.app.master_node master_node = self.app.master_node
if master_node is None: conn.answer(protocol.answerPrimaryMaster(master_node.getUUID(), []),
self.__notConnected(conn, packet)
else:
conn.answer(
protocol.answerPrimaryMaster(master_node.getUUID(), []),
packet.getId()) packet.getId())
class MasterEventHandler(EventHandler): class MasterEventHandler(EventHandler):
......
...@@ -101,8 +101,11 @@ class EventHandler(object): ...@@ -101,8 +101,11 @@ class EventHandler(object):
answer_packet = protocol.brokenNodeDisallowedError('go away') answer_packet = protocol.brokenNodeDisallowedError('go away')
conn.answer(answer_packet, packet.getId()) conn.answer(answer_packet, packet.getId())
conn.abort() conn.abort()
except NotReadyError: except NotReadyError, message:
conn.answer(protocol.notReady('retry later'), packet.getId()) if not message.args:
message = 'Retry Later'
message = str(message)
conn.answer(protocol.notReady(message), packet.getId())
conn.abort() conn.abort()
except ProtocolError, message: except ProtocolError, message:
message = str(message) message = str(message)
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment