Task #7432: Added support for LOFAR messages, and the protocols that we use

LCS/MessageBus/src/CMakeLists.txt

@@ -24,3 +24,4 @@ python_install(
   Message.py
   DESTINATION lofar/messagebus)
 
+add_subdirectory(Protocols)

LCS/MessageBus/src/Message.py

@@ -19,27 +19,25 @@
 import qpid.messaging
 import xml.dom.minidom as xml
 
-__slots__ = ["Message"]
-
 LOFAR_MSG_TEMPLATE = """
 <message>
    <header>
-      <system></system>
-      <version></version>
+      <system/>
+      <version/>
       <protocol>
-         <name></name>
-         <version></version>
+         <name/>
+         <version/>
       </protocol>
       <source>
-         <name></name>
-         <user></user>
-         <uuid></uuid>
-         <timestamp></timestamp>
-         <summary></summary>
+         <name/>
+         <user/>
+         <uuid/>
+         <timestamp/>
+         <summary/>
       </source>
       <ids>
-         <momid></momid>
-         <sasid></sasid>
+         <momid/>
+         <sasid/>
       </ids>
    </header>
    <payload>
@@ -48,7 +46,6 @@ LOFAR_MSG_TEMPLATE = """
 
 class Message(object):
     def __init__(self, from_, forUser, summary, protocol, protocolVersion, momid, sasid):
-      self.qpidMsg = qpid.messaging.Message()
       self.document = xml.parseString(LOFAR_MSG_TEMPLATE)
 
       for name, element in self._property_list().iteritems():
@@ -95,11 +92,6 @@ class Message(object):
 
     """ API (apart from properties). """
 
-    def setXMLPayload(self, payload):
-      self.qpidMsg.setContent(self.content_template % (payload,))
-
-    setTXTPayload = setXMLPayload
-
     def __repr__(self):
       return "Message(%s %s)" % (self.protocol, self.protocolVersion)
 
@@ -171,7 +163,11 @@ class Message(object):
     def _setXMLdata(self, name, data):
       return self._set_data(self._getXMLnode(name), data)
 
+    def qpidMsg(self):
+      qpidMsg = qpid.messaging.Message(self.document.toxml())
+
 if __name__ == "__main__":
   m = Message("FROM", "FORUSER", "SUMMARY", "PROTOCOL", "1.2.3", "11111", "22222")
   print str(m)
+  print m.document.toxml()
 

LCS/MessageBus/src/Protocols/CMakeLists.txt

+# $Id$
+
+include(PythonInstall)
+
+python_install(
+  __init__.py
+  DESTINATION lofar/messagebus/Protocols)

LCS/MessageBus/src/Protocols/TaskFeedbackDataproducts.py

+
+#!/usr/bin/python
+# Copyright (C) 2012-2015  ASTRON (Netherlands Institute for Radio Astronomy)
+# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
+#
+# This file is part of the LOFAR software suite.
+# The LOFAR software suite is free software: you can redistribute it and/or
+# modify it under the terms of the GNU General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# The LOFAR software suite is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
+
+import lofar.messagebus
+
+class TaskFeedbackDataproducts(lofar.messagebus.Message):
+  def __init__(self, from_, forUser, summary, momID, sasID, feedback):
+    super(TaskFeedbackDataproducts, self).__init__(
+      from_,
+      forUser,
+      summary,
+      "lofar.task.feedback.dataproducts",
+      "1.0.0",
+      momID,
+      sasID)
+
+    self.payload = feedback.toString()

LCS/MessageBus/src/Protocols/TaskFeedbackProcessing.py

+
+#!/usr/bin/python
+# Copyright (C) 2012-2015  ASTRON (Netherlands Institute for Radio Astronomy)
+# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
+#
+# This file is part of the LOFAR software suite.
+# The LOFAR software suite is free software: you can redistribute it and/or
+# modify it under the terms of the GNU General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# The LOFAR software suite is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
+
+import lofar.messagebus
+
+class TaskFeedbackProcessing(lofar.messagebus.Message):
+  def __init__(self, from_, forUser, summary, momID, sasID, feedback):
+    super(TaskFeedbackProcessing, self).__init__(
+      from_,
+      forUser,
+      summary,
+      "lofar.task.feedback.processing",
+      "1.0.0",
+      momID,
+      sasID)
+
+    self.payload = feedback.toString()

LCS/MessageBus/src/Protocols/TaskFeedbackStatus.py

+
+#!/usr/bin/python
+# Copyright (C) 2012-2015  ASTRON (Netherlands Institute for Radio Astronomy)
+# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
+#
+# This file is part of the LOFAR software suite.
+# The LOFAR software suite is free software: you can redistribute it and/or
+# modify it under the terms of the GNU General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# The LOFAR software suite is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
+
+#import lofar.messagebus.Message
+from Message import Message
+import xml.dom.minidom as xml
+
+LOFAR_STATUS_MSG_TEMPLATE = """
+<task>
+  <type/>
+  <state/>
+</task>"""
+
+class TaskFeedbackStatus(Message):
+  def __init__(self, from_, forUser, summary, momID, sasID, status):
+    super(TaskFeedbackStatus, self).__init__(
+      from_,
+      forUser,
+      summary,
+      "lofar.task.feedback.status",
+      "1.0.0",
+      momID,
+      sasID)
+
+    payload_document = xml.parseString(LOFAR_STATUS_MSG_TEMPLATE)
+
+    self._getXMLnode("message.payload").appendChild(payload_document.firstChild)
+
+    self.type_ = "pipeline"
+    self.state = "finished" if status else "aborted"
+
+  def _property_list(self):
+     properties = super(TaskFeedbackStatus, self)._property_list()
+     
+     properties.update( {
+       "type_": "message.payload.task.type",
+       "state": "message.payload.task.state",
+     } )
+
+     return properties
+
+if __name__ == "__main__":
+    msg = TaskFeedbackStatus("FROM", "FORUSER", "SUMMARY", "11111", "22222", True)
+    print msg.document.toxml()
+

LCS/MessageBus/src/Protocols/__init__.py

+# Copyright (C) 2012-2013  ASTRON (Netherlands Institute for Radio Astronomy)
+# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
+#
+# This file is part of the LOFAR software suite.
+# The LOFAR software suite is free software: you can redistribute it and/or
+# modify it under the terms of the GNU General Public License as published
+# by the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# The LOFAR software suite is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along
+# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
+#
+# $Id$