Skip to content

N
neoppod
Commit 3774858d authored by Grégory Wisniewski's avatar Grégory Wisniewski
Browse files
Options

Factorise struct.error exception handling with a decorator over decoding 

functions. Factorise check of enum based values and string decoding. 
Unify tuple unpacking by always using parenthesis.


git-svn-id: https://svn.erp5.org/repos/neo/branches/prototype3@663 71dcc9de-d417-0410-9af5-da40c76e7ee4
parent 193a31e1
Show whitespace changes
Inline Side-by-side
Showing with 304 additions and 409 deletions
neo/protocol.py
View file @ 3774858d
......@@ -453,135 +453,149 @@ def parse(msg):
packet.setId(msg_id)
return packet
# packet decoding
def _decodeError(body):
def handle_errors(decoder):
""" Decorator to be used on encoding/decoding methods. Intercept struct
(pack/unpack) exceptions and wrap them in PacketMalformedError """
def wrapper(body):
try:
body = body
code, size = unpack('!HL', body[:6])
message = body[6:]
return decoder(body)
except struct.error, msg:
raise PacketMalformedError('invalid error message')
if len(message) != size:
raise PacketMalformedError('invalid error message size')
return code, message
name = decoder.__name__
raise PacketMalformedError("%s fail (%s)" % (name, msg))
except PacketMalformedError, msg:
name = decoder.__name__
raise PacketMalformedError("%s fail (%s)" % (name, msg))
return wrapper
def _checkClusterState(state):
cluster_state = cluster_states.get(state)
if cluster_state is None:
raise PacketMalformedError('invalid node state %d' % state)
return cluster_state
def _checkNodeState(state):
node_state = node_states.get(state)
if node_state is None:
raise PacketMalformedError('invalid node state %d' % state)
return node_state
def _checkNodeType(type):
node_type = node_types.get(type)
if node_type is None:
raise PacketMalformedError('invalide node type %d' % type)
return node_type
def _readString(buffer, name, offset=0):
buffer = buffer[offset:]
(size, ) = unpack('!L', buffer[:4])
string = buffer[4:4+size]
if len(string) != size:
raise PacketMalformedError("can't read string <%s>" % name)
return (string, buffer[offset+size:])
# packet decoding
@handle_errors
def _decodeError(body):
(code, ) = unpack('!H', body[:2])
(message, _) = _readString(body, 'message', offset=2)
return (code, message)
decode_table[ERROR] = _decodeError
@handle_errors
def _decodePing(body):
pass
decode_table[PING] = _decodePing
@handle_errors
def _decodePong(body):
pass
decode_table[PONG] = _decodePong
@handle_errors
def _decodeRequestNodeIdentification(body):
try:
body = body
major, minor, node_type, uuid, ip_address, port, size \
= unpack('!LLH16s4sHL', body[:36])
r = unpack('!LLH16s4sH', body[:32])
major, minor, node_type, uuid, ip_address, port = r
ip_address = inet_ntoa(ip_address)
name = body[36:]
except struct.error, msg:
raise PacketMalformedError('invalid request node identification')
if size != len(name):
raise PacketMalformedError('invalid name size')
node_type = node_types.get(node_type)
if node_type not in VALID_NODE_TYPE_LIST:
raise PacketMalformedError('invalid node type %d' % node_type)
(name, _) = _readString(body, 'name', offset=32)
node_type = _checkNodeType(node_type)
if (major, minor) != PROTOCOL_VERSION:
raise PacketMalformedError('protocol version mismatch')
return node_type, uuid, ip_address, port, name
decode_table[REQUEST_NODE_IDENTIFICATION] = _decodeRequestNodeIdentification
@handle_errors
def _decodeAcceptNodeIdentification(body):
try:
node_type, uuid, ip_address, port, num_partitions, num_replicas, your_uuid \
= unpack('!H16s4sHLL16s', body)
r = unpack('!H16s4sHLL16s', body)
node_type, uuid, ip_address, port, num_partitions, num_replicas, your_uuid = r
ip_address = inet_ntoa(ip_address)
except struct.error, msg:
raise PacketMalformedError('invalid accept node identification')
node_type = node_types.get(node_type)
if node_type not in VALID_NODE_TYPE_LIST:
raise PacketMalformedError('invalid node type %d' % node_type)
return node_type, uuid, ip_address, port, num_partitions, num_replicas, your_uuid
node_type = _checkNodeType(node_type)
return (node_type, uuid, ip_address, port, num_partitions, num_replicas, your_uuid)
decode_table[ACCEPT_NODE_IDENTIFICATION] = _decodeAcceptNodeIdentification
@handle_errors
def _decodeAskPrimaryMaster(body):
pass
decode_table[ASK_PRIMARY_MASTER] = _decodeAskPrimaryMaster
@handle_errors
def _decodeAnswerPrimaryMaster(body):
try:
primary_uuid, n = unpack('!16sL', body[:20])
(primary_uuid, n) = unpack('!16sL', body[:20])
known_master_list = []
for i in xrange(n):
ip_address, port, uuid = unpack('!4sH16s', body[20+i*22:42+i*22])
ip_address = inet_ntoa(ip_address)
known_master_list.append((ip_address, port, uuid))
except struct.error, msg:
raise PacketMalformedError('invalid answer primary master')
return primary_uuid, known_master_list
return (primary_uuid, known_master_list)
decode_table[ANSWER_PRIMARY_MASTER] = _decodeAnswerPrimaryMaster
@handle_errors
def _decodeAnnouncePrimaryMaster(body):
pass
decode_table[ANNOUNCE_PRIMARY_MASTER] = _decodeAnnouncePrimaryMaster
@handle_errors
def _decodeReelectPrimaryMaster(body):
pass
decode_table[REELECT_PRIMARY_MASTER] = _decodeReelectPrimaryMaster
@handle_errors
def _decodeNotifyNodeInformation(body):
try:
n = unpack('!L', body[:4])[0]
(n,) = unpack('!L', body[:4])
node_list = []
for i in xrange(n):
r = unpack('!H4sH16sH', body[4+i*26:30+i*26])
node_type, ip_address, port, uuid, state = r
ip_address = inet_ntoa(ip_address)
node_type = node_types.get(node_type)
if node_type not in VALID_NODE_TYPE_LIST:
raise PacketMalformedError('invalid node type %d' % node_type)
state = node_states.get(state)
if state not in VALID_NODE_STATE_LIST:
raise PacketMalformedError('invalid node state %d' % state)
node_type = _checkNodeType(node_type)
state = _checkNodeState(state)
node_list.append((node_type, ip_address, port, uuid, state))
except PacketMalformedError:
raise
except struct.error, msg:
raise PacketMalformedError('invalid answer node information')
return (node_list,)
decode_table[NOTIFY_NODE_INFORMATION] = _decodeNotifyNodeInformation
@handle_errors
def _decodeAskLastIDs(body):
pass
decode_table[ASK_LAST_IDS] = _decodeAskLastIDs
@handle_errors
def _decodeAnswerLastIDs(body):
try:
loid, ltid, lptid = unpack('!8s8s8s', body)
except struct.error, msg:
raise PacketMalformedError('invalid answer last ids')
return loid, ltid, lptid
return unpack('!8s8s8s', body) # (loid, ltid, lptid)
decode_table[ANSWER_LAST_IDS] = _decodeAnswerLastIDs
@handle_errors
def _decodeAskPartitionTable(body):
try:
n = unpack('!L', body[:4])[0]
(n,) = unpack('!L', body[:4])
offset_list = []
for i in xrange(n):
offset = unpack('!L', body[4+i*4:8+i*4])[0]
offset_list.append(offset)
except struct.error, msg:
raise PacketMalformedError('invalid ask partition table')
return (offset_list,)
decode_table[ASK_PARTITION_TABLE] = _decodeAskPartitionTable
@handle_errors
def _decodeAnswerPartitionTable(body):
try:
ptid, n = unpack('!8sL', body[:12])
index = 12
(ptid, n) = unpack('!8sL', body[:index])
row_list = []
cell_list = []
for i in xrange(n):
......@@ -594,15 +608,13 @@ def _decodeAnswerPartitionTable(body):
cell_list.append((uuid, state))
row_list.append((offset, tuple(cell_list)))
del cell_list[:]
except struct.error, msg:
raise PacketMalformedError('invalid answer partition table')
return ptid, row_list
return (ptid, row_list)
decode_table[ANSWER_PARTITION_TABLE] = _decodeAnswerPartitionTable
@handle_errors
def _decodeSendPartitionTable(body):
try:
ptid, n = unpack('!8sL', body[:12])
index = 12
(ptid, n,) = unpack('!8sL', body[:index])
row_list = []
cell_list = []
for i in xrange(n):
......@@ -615,345 +627,263 @@ def _decodeSendPartitionTable(body):
cell_list.append((uuid, state))
row_list.append((offset, tuple(cell_list)))
del cell_list[:]
except struct.error, msg:
raise PacketMalformedError('invalid send partition table')
return ptid, row_list
return (ptid, row_list)
decode_table[SEND_PARTITION_TABLE] = _decodeSendPartitionTable
@handle_errors
def _decodeNotifyPartitionChanges(body):
try:
ptid, n = unpack('!8sL', body[:12])
(ptid, n) = unpack('!8sL', body[:12])
cell_list = []
for i in xrange(n):
offset, uuid, state = unpack('!L16sH', body[12+i*22:34+i*22])
(offset, uuid, state) = unpack('!L16sH', body[12+i*22:34+i*22])
state = partition_cell_states.get(state)
cell_list.append((offset, uuid, state))
except struct.error, msg:
raise PacketMalformedError('invalid notify partition changes')
return ptid, cell_list
decode_table[NOTIFY_PARTITION_CHANGES] = _decodeNotifyPartitionChanges
@handle_errors
def _decodeStartOperation(body):
pass
decode_table[START_OPERATION] = _decodeStartOperation
@handle_errors
def _decodeStopOperation(body):
pass
decode_table[STOP_OPERATION] = _decodeStopOperation
@handle_errors
def _decodeAskUnfinishedTransactions(body):
pass
decode_table[ASK_UNFINISHED_TRANSACTIONS] = _decodeAskUnfinishedTransactions
@handle_errors
def _decodeAnswerUnfinishedTransactions(body):
try:
n = unpack('!L', body[:4])[0]
(n,) = unpack('!L', body[:4])
tid_list = []
for i in xrange(n):
tid = unpack('8s', body[4+i*8:12+i*8])[0]
tid_list.append(tid)
except struct.error, msg:
raise PacketMalformedError('invalid answer unfinished transactions')
return (tid_list,)
decode_table[ANSWER_UNFINISHED_TRANSACTIONS] = _decodeAnswerUnfinishedTransactions
@handle_errors
def _decodeAskObjectPresent(body):
try:
oid, tid = unpack('8s8s', body)
except struct.error, msg:
raise PacketMalformedError('invalid ask object present')
return oid, tid
return unpack('8s8s', body) # oid, tid
decode_table[ASK_OBJECT_PRESENT] = _decodeAskObjectPresent
@handle_errors
def _decodeAnswerObjectPresent(body):
try:
oid, tid = unpack('8s8s', body)
except struct.error, msg:
raise PacketMalformedError('invalid answer object present')
return oid, tid
return unpack('8s8s', body) # oid, tid
decode_table[ANSWER_OBJECT_PRESENT] = _decodeAnswerObjectPresent
@handle_errors
def _decodeDeleteTransaction(body):
try:
tid = unpack('8s', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid delete transaction')
return (tid,)
return unpack('8s', body) # tid
decode_table[DELETE_TRANSACTION] = _decodeDeleteTransaction
@handle_errors
def _decodeCommitTransaction(body):
try:
tid = unpack('8s', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid commit transaction')
return (tid,)
return unpack('8s', body) # tid
decode_table[COMMIT_TRANSACTION] = _decodeCommitTransaction
@handle_errors
def _decodeAskNewTID(body):
pass
decode_table[ASK_NEW_TID] = _decodeAskNewTID
@handle_errors
def _decodeAnswerNewTID(body):
try:
tid = unpack('8s', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid answer new tid')
return (tid,)
return unpack('8s', body) # tid
decode_table[ANSWER_NEW_TID] = _decodeAnswerNewTID
@handle_errors
def _decodeAskNewOIDs(body):
try:
num_oids = unpack('!H', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid ask new oids')
return (num_oids,)
return unpack('!H', body) # num oids
decode_table[ASK_NEW_OIDS] = _decodeAskNewOIDs
@handle_errors
def _decodeAnswerNewOIDs(body):
try:
n = unpack('!H', body[:2])[0]
(n,) = unpack('!H', body[:2])
oid_list = []
for i in xrange(n):
oid = unpack('8s', body[2+i*8:10+i*8])[0]
oid_list.append(oid)
except struct.error, msg:
raise PacketMalformedError('invalid answer new oids')
return (oid_list,)
decode_table[ANSWER_NEW_OIDS] = _decodeAnswerNewOIDs
@handle_errors
def _decodeFinishTransaction(body):
try:
tid, n = unpack('!8sL', body[:12])
(tid, n) = unpack('!8sL', body[:12])
oid_list = []
for i in xrange(n):
oid = unpack('8s', body[12+i*8:20+i*8])[0]
oid_list.append(oid)
except struct.error, msg:
raise PacketMalformedError('invalid finish transaction')
return oid_list, tid
return (oid_list, tid)
decode_table[FINISH_TRANSACTION] = _decodeFinishTransaction
@handle_errors
def _decodeNotifyTransactionFinished(body):
try:
tid = unpack('8s', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid notify transactin finished')
return (tid,)
return unpack('8s', body) # tid
decode_table[NOTIFY_TRANSACTION_FINISHED] = _decodeNotifyTransactionFinished
@handle_errors
def _decodeLockInformation(body):
try:
tid = unpack('8s', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid lock information')
return (tid,)
return unpack('8s', body) # tid
decode_table[LOCK_INFORMATION] = _decodeLockInformation
@handle_errors
def _decodeNotifyInformationLocked(body):
try:
tid = unpack('8s', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid notify information locked')
return (tid,)
return unpack('8s', body) # tid
decode_table[NOTIFY_INFORMATION_LOCKED] = _decodeNotifyInformationLocked
@handle_errors
def _decodeInvalidateObjects(body):
try:
tid, n = unpack('!8sL', body[:12])
(tid, n) = unpack('!8sL', body[:12])
oid_list = []
for i in xrange(12, 12 + n * 8, 8):
oid = unpack('8s', body[i:i+8])[0]
oid_list.append(oid)
except struct.error, msg:
raise PacketMalformedError('invalid finish transaction')
return oid_list, tid
return (oid_list, tid)
decode_table[INVALIDATE_OBJECTS] = _decodeInvalidateObjects
@handle_errors
def _decodeUnlockInformation(body):
try:
tid = unpack('8s', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid unlock information')
return (tid,)
return unpack('8s', body) # tid
decode_table[UNLOCK_INFORMATION] = _decodeUnlockInformation
@handle_errors
def _decodeAbortTransaction(body):
try:
tid = unpack('8s', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid abort transaction')
return (tid,)
return unpack('8s', body) # tid
decode_table[ABORT_TRANSACTION] = _decodeAbortTransaction
@handle_errors
def _decodeAskStoreObject(body):
try:
oid, serial, tid, compression, checksum, data_len \
= unpack('!8s8s8sBLL', body[:33])
data = body[33:]
except struct.error, msg:
raise PacketMalformedError('invalid ask store object')
if data_len != len(data):
raise PacketMalformedError('invalid data size')
return oid, serial, compression, checksum, data, tid
r = unpack('!8s8s8sBL', body[:29])
oid, serial, tid, compression, checksum = r
(data, _) = _readString(body, 'data', offset=29)
return (oid, serial, compression, checksum, data, tid)
decode_table[ASK_STORE_OBJECT] = _decodeAskStoreObject
@handle_errors
def _decodeAnswerStoreObject(body):
try:
conflicting, oid, serial = unpack('!B8s8s', body)
except struct.error, msg:
raise PacketMalformedError('invalid answer store object')
return conflicting, oid, serial
return unpack('!B8s8s', body) # conflicting, oid, serial
decode_table[ANSWER_STORE_OBJECT] = _decodeAnswerStoreObject
@handle_errors
def _decodeAskStoreTransaction(body):
try:
tid, oid_len, user_len, desc_len, ext_len \
= unpack('!8sLHHH', body[:18])
offset = 18
user = body[offset:offset+user_len]
offset += user_len
desc = body[offset:offset+desc_len]
offset += desc_len
ext = body[offset:offset+ext_len]
offset += ext_len
r = unpack('!8sLHHH', body[:18])
tid, oid_len, user_len, desc_len, ext_len = r
body = body[18:]
user = body[:user_len]
body = body[user_len:]
desc = body[:desc_len]
body = body[desc_len:]
ext = body[:ext_len]
body = body[ext_len:]
oid_list = []
for i in xrange(oid_len):
oid = unpack('8s', body[offset:offset+8])[0]
offset += 8
(oid, ) = unpack('8s', body[:8])
body = body[8:]
oid_list.append(oid)
except struct.error, msg:
raise PacketMalformedError('invalid ask store transaction')
return tid, user, desc, ext, oid_list
return (tid, user, desc, ext, oid_list)
decode_table[ASK_STORE_TRANSACTION] = _decodeAskStoreTransaction
@handle_errors
def _decodeAnswerStoreTransaction(body):
try:
tid = unpack('8s', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid answer store transaction')
return (tid,)
return unpack('8s', body) # tid
decode_table[ANSWER_STORE_TRANSACTION] = _decodeAnswerStoreTransaction
@handle_errors
def _decodeAskObject(body):
try:
oid, serial, tid = unpack('8s8s8s', body)
except struct.error, msg:
raise PacketMalformedError('invalid ask object')
return oid, serial, tid
return unpack('8s8s8s', body) # oid, serial, tid
decode_table[ASK_OBJECT] = _decodeAskObject
@handle_errors
def _decodeAnswerObject(body):
try:
oid, serial_start, serial_end, compression, checksum, data_len \
= unpack('!8s8s8sBLL', body[:33])
data = body[33:]
except struct.error, msg:
raise PacketMalformedError('invalid answer object')
if len(data) != data_len:
raise PacketMalformedError('invalid data size')
return oid, serial_start, serial_end, compression, checksum, data
r = unpack('!8s8s8sBL', body[:29])
oid, serial_start, serial_end, compression, checksum = r
(data, _) = _readString(body, 'data', offset=29)
return (oid, serial_start, serial_end, compression, checksum, data)
decode_table[ANSWER_OBJECT] = _decodeAnswerObject
@handle_errors
def _decodeAskTIDs(body):
try:
first, last, partition = unpack('!QQL', body)
except struct.error, msg:
raise PacketMalformedError('invalid ask tids')
return first, last, partition
return unpack('!QQL', body) # first, last, partition
decode_table[ASK_TIDS] = _decodeAskTIDs
@handle_errors
def _decodeAnswerTIDs(body):
try:
n = unpack('!L', body[:4])[0]
(n, ) = unpack('!L', body[:4])
tid_list = []
for i in xrange(n):
tid = unpack('8s', body[4+i*8:12+i*8])[0]
tid_list.append(tid)
except struct.error, msg:
raise PacketMalformedError('invalid answer tids')
return (tid_list,)
decode_table[ANSWER_TIDS] = _decodeAnswerTIDs
@handle_errors
def _decodeAskTransactionInformation(body):
try:
tid = unpack('8s', body)[0]
except struct.error, msg:
raise PacketMalformedError('invalid ask transaction information')
return (tid,)
return unpack('8s', body) # tid
decode_table[ASK_TRANSACTION_INFORMATION] = _decodeAskTransactionInformation
@handle_errors
def _decodeAnswerTransactionInformation(body):
try:
tid, user_len, desc_len, ext_len, oid_len \
= unpack('!8sHHHL', body[:18])
offset = 18
user = body[offset:offset+user_len]
offset += user_len
desc = body[offset:offset+desc_len]
offset += desc_len
ext = body[offset:offset+ext_len]
offset += ext_len
r = unpack('!8sHHHL', body[:18])
tid, user_len, desc_len, ext_len, oid_len = r
body = body[18:]
user = body[:user_len]
body = body[user_len:]
desc = body[:desc_len]
body = body[desc_len:]
ext = body[:ext_len]
body = body[ext_len:]
oid_list = []
for i in xrange(oid_len):
oid = unpack('8s', body[offset+i*8:offset+8+i*8])[0]
(oid, ) = unpack('8s', body[:8])
body = body[8:]
oid_list.append(oid)
except struct.error, msg:
raise PacketMalformedError('invalid answer transaction information')
return tid, user, desc, ext, oid_list
return (tid, user, desc, ext, oid_list)
decode_table[ANSWER_TRANSACTION_INFORMATION] = _decodeAnswerTransactionInformation
@handle_errors
def _decodeAskObjectHistory(body):
try:
oid, first, last = unpack('!8sQQ', body)
except struct.error, msg:
raise PacketMalformedError('invalid ask object history')
return oid, first, last
return unpack('!8sQQ', body) # oid, first, last
decode_table[ASK_OBJECT_HISTORY] = _decodeAskObjectHistory
@handle_errors
def _decodeAnswerObjectHistory(body):
try:
oid, length = unpack('!8sL', body[:12])
(oid, length) = unpack('!8sL', body[:12])
history_list = []
for i in xrange(12, 12 + length * 12, 12):
serial, size = unpack('!8sL', body[i:i+12])
history_list.append((serial, size))
except struct.error, msg:
raise PacketMalformedError('invalid answer object history')
return oid, history_list
return (oid, history_list)
decode_table[ANSWER_OBJECT_HISTORY] = _decodeAnswerObjectHistory
@handle_errors
def _decodeAskOIDs(body):
try:
first, last, partition = unpack('!QQL', body)
except struct.error, msg:
raise PacketMalformedError('invalid ask oids')
return first, last, partition
return unpack('!QQL', body) # first, last, partition
decode_table[ASK_OIDS] = _decodeAskOIDs
@handle_errors
def _decodeAnswerOIDs(body):
try:
n = unpack('!L', body[:4])[0]
(n,) = unpack('!L', body[:4])
oid_list = []
for i in xrange(n):
oid = unpack('8s', body[4+i*8:12+i*8])[0]
oid_list.append(oid)
except struct.error, msg:
raise PacketMalformedError('invalid answer oids')
return (oid_list,)
decode_table[ANSWER_OIDS] = _decodeAnswerOIDs
@handle_errors
def _decodeAskPartitionList(body):
try:
min_offset, max_offset, uuid = unpack('!LL16s', body)
except struct.error, msg:
raise PacketMalformedError('invalid ask partition list')
return (min_offset, max_offset, uuid)
return unpack('!LL16s', body) # min_offset, max_offset, uuid
decode_table[ASK_PARTITION_LIST] = _decodeAskPartitionList
@handle_errors
def _decodeAnswerPartitionList(body):
try:
ptid, n = unpack('!8sL', body[:12])
index = 12
(ptid, n) = unpack('!8sL', body[:index])
row_list = []
cell_list = []
for i in xrange(n):
......@@ -966,105 +896,70 @@ def _decodeAnswerPartitionList(body):
cell_list.append((uuid, state))
row_list.append((offset, tuple(cell_list)))
del cell_list[:]
except struct.error, msg:
raise PacketMalformedError('invalid answer partition list')
return ptid, row_list
return (ptid, row_list)
decode_table[ANSWER_PARTITION_LIST] = _decodeAnswerPartitionList
@handle_errors
def _decodeAskNodeList(body):
try:
node_type = unpack('!H', body)[0]
node_type = node_types.get(node_type)
if node_type not in VALID_NODE_TYPE_LIST:
raise PacketMalformedError('invalid node type %d' % node_type)
except struct.error, msg:
raise PacketMalformedError('invalid ask node list')
(node_type, ) = unpack('!H', body)
node_type = _checkNodeType(node_type)
return (node_type,)
decode_table[ASK_NODE_LIST] = _decodeAskNodeList
@handle_errors
def _decodeAnswerNodeList(body):
try:
n = unpack('!L', body[:4])[0]
(n,) = unpack('!L', body[:4])
node_list = []
for i in xrange(n):
r = unpack('!H4sH16sH', body[4+i*26:30+i*26])
node_type, ip_address, port, uuid, state = r
ip_address = inet_ntoa(ip_address)
node_type = node_types.get(node_type)
if node_type not in VALID_NODE_TYPE_LIST:
raise PacketMalformedError('invalid node type %d' % node_type)
state = node_states.get(state)
if state not in VALID_NODE_STATE_LIST:
raise PacketMalformedError('invalid node state %d' % state)
node_type = _checkNodeType(node_type)
state = checkNodeState(state)
node_list.append((node_type, ip_address, port, uuid, state))
except struct.error, msg:
raise PacketMalformedError('invalid answer node information')
return (node_list,)
decode_table[ANSWER_NODE_LIST] = _decodeAnswerNodeList
@handle_errors
def _decodeSetNodeState(body):
try:
uuid, state, modify = unpack('!16sHB', body)
state = node_states.get(state)
if state not in VALID_NODE_STATE_LIST:
raise PacketMalformedError('invalid node state %d' % state)
except struct.error, msg:
raise PacketMalformedError('invalid set node state')
(uuid, state, modify) = unpack('!16sHB', body)
state = checkNodeState(state)
return (uuid, state, modify)
decode_table[SET_NODE_STATE] = _decodeSetNodeState
@handle_errors
def _decodeAnswerNodeState(body):
try:
uuid, state = unpack('!16sH', body)
state = node_states.get(state)
if state not in VALID_NODE_STATE_LIST:
raise PacketMalformedError('invalid node state %d' % state)
except struct.error, msg:
raise PacketMalformedError('invalid answer node state')
(uuid, state) = unpack('!16sH', body)
state = checkNodeState(state)
return (uuid, state)
decode_table[ANSWER_NODE_STATE] = _decodeAnswerNodeState
@handle_errors
def _decodeSetClusterState(body):
try:
state, len_name = unpack('!HL', body[:6])
name = body[6:]
if len_name != len(name):
raise PacketMalformedError('invalid name size')
state = cluster_states.get(state)
if state is None:
raise PacketMalformedError('invalid cluster state %d' % state)
except struct.error, msg:
raise PacketMalformedError('invalid set node state')
(state, ) = unpack('!H', body[:2])
(name, _) = _readString(body, 'name', offset=2)
state = _checkClusterState(state)
return (name, state)
decode_table[SET_CLUSTER_STATE] = _decodeSetClusterState
@handle_errors
def _decodeAnswerClusterState(body):
try:
(state, ) = unpack('!H', body)
state = cluster_states.get(state)
if state is None:
raise PacketMalformedError('invalid cluster state %d' % state)
except struct.error, msg:
raise PacketMalformedError('invalid answer cluster state')
state = _checkClusterState(state)
return (state, )
decode_table[ANSWER_CLUSTER_STATE] = _decodeAnswerClusterState
@handle_errors
def _decodeAddPendingNodes(body):
try:
(n, ) = unpack('!H', body[:2])
uuid_list = [unpack('!16s', body[2+i*16:18+i*16])[0] for i in xrange(n)]
except struct.error, msg:
raise PacketMalformedError('invalid add pending nodes')
return (uuid_list, )
decode_table[ADD_PENDING_NODES] = _decodeAddPendingNodes
@handle_errors
def _decodeAnswerNewNodes(body):
try:
(n, ) = unpack('!H', body[:2])
uuid_list = [unpack('!16s', body[2+i*16:18+i*16])[0] for i in xrange(n)]
except struct.error, msg:
raise PacketMalformedError('invalid answer new nodes')
return (uuid_list, )
decode_table[ANSWER_NEW_NODES] = _decodeAnswerNewNodes
......
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
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7