Integrating Message Infrastructure into Zambeze

[JoshuaSBrown]

First step, in integrating the messaging infrastructure into Zambeze.

[JoshuaSBrown]

@tskluzac here I have tried to demonstrate the use of the template idea that you proposed, doesn't work to well right now because we are returning a dict where the user can just add arbitrary keys on accident. I have been talking with @renan-souza about this. It would be ideal if the template had predefined keys that were immutable this would ensure consistency. If you know a good way to do this it would be appreciated.

[tskluzac]

What is the difference between immutable_msg and immutable_msg_move? Unclear from variable names + docstring.

[JoshuaSBrown]

The first is doing a transfer between Globus imports the second is moving files from and enpoint to somewhere else on the POSIX file system. I should probably add some comments, or do you have some ideas for better variable names?

[tskluzac]

Curious what this is doing. It looks like it's 'muting' the second element of a list. Is there a cleaner way of doing this?

[JoshuaSBrown]

msg_template_move is a tuple the first element is the MessageType, and the second element is a dict which contains the message envelope as defined in the detailed design document. Do you have any suggestions?

[JoshuaSBrown]

Maybe we could have a template object that is returned.

class MessageTemplate:
 
 def __init__(self):
    self.__type = MessageType.ACTIVITY
    self.__message = { ...Template... }

 @property
  def type(self) -> MessageType:
    return self.__type
  
  @property
  def data(self):
    return self.__message

[tskluzac]

Generally the "+" string concatenation is frowned upon due to unpredictable behavior. Could we use f-strings or even URL/Path-join?

[JoshuaSBrown]

Good catch, I did not know that. Yes.

[tskluzac]

Repeat comment from earlier (are we editing second element of list? and can this be done cleaner)

[JoshuaSBrown]

The following changes have occurred in this PR. The PluginMessageHelper class was split into two separate classes to follow the separation of concerns principle and the single responsibility principle. Red indicates removal Blue indicates a new addition.

[JoshuaSBrown]

Ideally the only thing you need to work with in this PR are the MessageFactory, the AbstractMessage and the template.

# Globus example
plugins = Plugins()
msg_factory = MessageFactory(plugins)
activity_template = msg_factory.createTemplate(
        MessageType.ACTIVITY,
        ActivityType.PLUGIN,
        args={"plugin": "globus", "action": "transfer"},
)
activity_template[1].message_id = str(uuid.uuid4())
activity_template[1].activity_id = str(uuid.uuid4())
activity_template[1].agent_id = str(uuid.uuid4())
activity_template[1].campaign_id = str(uuid.uuid4())
activity_template[1].credential = {}
activity_template[1].submission_time = str(int(time.time()))
activity_template[1].body.parameters.transfer.items[0].source = ""
activity_template[1].body.parameters.transfer.items[0].destination = ""
activity_template[1].needs = []

# Will check that the required message values are filled
msg = factory.create(activity_template)

# Will check that the msg can be executed on local machine
results = plugins.check(msg)

# Will run the activity
plugins.run(msg)

This behavior has been added and integrated both in the processor and the agent files.

[JoshuaSBrown]

@wilkinson This is ready for your input. @tskluzac I have tried to address your concerns here as well.