jp (pubsub/transform): use new psItemsSend method, it is not needed anymore to send items one by one when --admin is not used

This commit is contained in:
Goffi 2019-02-15 22:13:43 +01:00
commit d747c46736

View file

@ -1552,26 +1552,17 @@ class Transform(base.CommandBase):
u'Return "DELETE" string to delete the item, and "SKIP" to ignore it'),
)
def psAdminItemsSendCb(self, item_ids, metadata):
self.disp(_(u'items published with ids {item_ids}').format(
item_ids=u', '.join(item_ids)))
def psItemsSendCb(self, item_ids, metadata):
if item_ids:
self.disp(_(u'items published with ids {item_ids}').format(
item_ids=u', '.join(item_ids)))
else:
self.disp(_(u'items published'))
if self.args.all:
return self.handleNextPage(metadata)
else:
self.host.quit()
def psItemsSendCb(self, item_id, metadata, show_mess=True):
if show_mess:
self.disp(u'item published with id {item_id}'.format(item_id=item_id))
if self.items_sent == self.items_to_send:
if self.args.all:
return self.handleNextPage(metadata)
self.disp(u'all items published')
self.host.quit()
def psRetractItemCb(self, item_id, metadata):
self.psItemsSendCb(item_id, metadata, show_mess=False)
def handleNextPage(self, metadata):
"""Retrieve new page through RSM or quit if we're in the last page
@ -1621,11 +1612,7 @@ class Transform(base.CommandBase):
def psItemsGetCb(self, ps_result):
items, metadata = ps_result
if self.args.admin:
new_items = []
else:
self.items_to_send = len(items)
self.items_sent = 0
new_items = []
for item in items:
if self.check_duplicates:
@ -1655,8 +1642,6 @@ class Transform(base.CommandBase):
self.disp(u"The command returned a non zero status while parsing the "
u"following item:\n\n{item}".format(item=item), error=True)
if self.args.ignore_errors:
if not self.args.admin:
self.items_to_send -= 1
continue
else:
self.host.quit(C.EXIT_CMD_ERROR)
@ -1669,19 +1654,20 @@ class Transform(base.CommandBase):
item_id = item_elt.get('id')
self.disp(_(u"Deleting item {item_id}").format(item_id=item_id))
if self.args.apply:
if not self.args.admin:
# we need to increase the counter as if the item were re-published
self.items_sent += 1
# FIXME: we don't wait for item to be retracted which can cause
# trouble in case of error just before the end of the command
# (the error message may be missed).
# Once moved to Python 3, we must wait for it by using a
# coroutine.
self.host.bridge.psRetractItem(
self.args.service,
self.args.node,
item_id,
False,
self.profile,
callback=partial(self.psRetractItemCb, metadata=metadata),
errback=partial(
self.errback,
msg=_(u"can't delete item: {}"),
msg=_(u"can't delete item [%s]: {}" % item_id),
exit_code=C.EXIT_BRIDGE_ERRBACK,
),
)
@ -1690,11 +1676,6 @@ class Transform(base.CommandBase):
item_elt, __ = xml_tools.etreeParse(self, item)
item_id = item_elt.get('id')
self.disp(_(u"Skipping item {item_id}").format(item_id=item_id))
if self.args.apply:
if not self.args.admin:
# see above
self.items_sent += 1
self.psItemsSendCb(item_id, metadata, show_mess=False)
continue
element, etree = xml_tools.etreeParse(self, cmd_std_out)
@ -1710,51 +1691,42 @@ class Transform(base.CommandBase):
pretty_print=True)
self.disp(serialised)
else:
# we will apply the change, either in admin request or as a simple
# pubsub one
if self.args.admin:
new_items.append(etree.tostring(element, encoding="unicode"))
else:
# there is currently no method to send several items at once
# so we publish them one by one
payload = etree.tostring(xml_tools.getPayload(self, element),
encoding="unicode")
item_id = element.get(u'id', '')
self.host.bridge.psItemSend(
self.args.service,
self.args.node,
payload,
item_id,
{},
self.profile,
callback=partial(self.psItemsSendCb, metadata=metadata),
errback=partial(
self.errback,
msg=_(u"can't send item: {}"),
exit_code=C.EXIT_BRIDGE_ERRBACK,
),
)
self.items_sent += 1
new_items.append(etree.tostring(element, encoding="unicode"))
if not self.args.apply:
# on dry run we have nothing to wait for, we can quit
if self.args.all:
return self.handleNextPage(metadata)
self.host.quit()
elif self.args.admin:
self.host.bridge.psAdminItemsSend(
self.args.service,
self.args.node,
new_items,
u"",
self.profile,
callback=partial(self.psAdminItemsSendCb, metadata=metadata),
errback=partial(
self.errback,
msg=_(u"can't send item: {}"),
exit_code=C.EXIT_BRIDGE_ERRBACK,
),
)
else:
if self.args.admin:
self.host.bridge.psAdminItemsSend(
self.args.service,
self.args.node,
new_items,
u"",
self.profile,
callback=partial(self.psItemsSendCb, metadata=metadata),
errback=partial(
self.errback,
msg=_(u"can't send item: {}"),
exit_code=C.EXIT_BRIDGE_ERRBACK,
),
)
else:
self.host.bridge.psItemsSend(
self.args.service,
self.args.node,
new_items,
u"",
self.profile,
callback=partial(self.psItemsSendCb, metadata=metadata),
errback=partial(
self.errback,
msg=_(u"can't send item: {}"),
exit_code=C.EXIT_BRIDGE_ERRBACK,
),
)
def start(self):
if self.args.all and self.args.order_by != C.ORDER_BY_CREATION: