Task #7573: Allow compilation and using the MessageBus classes even if QPID is...

Task #7573: Allow compilation and using the MessageBus classes even if QPID is not installed, through the use of a fallback that does not actually send or receive messages.

CEP/Pipeline/recipes/CMakeLists.txt

@@ -2,4 +2,8 @@
 
 lofar_package(Pipeline-Recipes 0.1)
 
+# The pipeline.cfg needs to know whether QPID is installed
+include(LofarFindPackage)
+lofar_find_package(QPID)
+
 add_subdirectory(sip)

CEP/Pipeline/recipes/sip/CMakeLists.txt

@@ -96,6 +96,13 @@ install(FILES
   ${CMAKE_CURRENT_BINARY_DIR}/tasks.cfg
   DESTINATION share/pipeline)
 
+# PIPELINE_FEEDBACK_METHOD is used in pipeline.cfg.in to enable/disable qpid feedback
+if(HAVE_QPID)
+  set(PIPELINE_FEEDBACK_METHOD "messagebus")
+else(HAVE_QPID)
+  set(PIPELINE_FEEDBACK_METHOD "none")
+endif(HAVE_QPID)
+
 configure_file(
   ${CMAKE_CURRENT_SOURCE_DIR}/pipeline.cfg.in
   ${CMAKE_CURRENT_BINARY_DIR}/pipeline.cfg)

CEP/Pipeline/recipes/sip/pipeline.cfg.in

@@ -29,4 +29,4 @@ xml_stat_file = %(runtime_directory)s/%(job_name)s/logs/%(start_time)s/statistic
 # Valid options:
 #    messagebus    Send feedback and status using LCS/MessageBus
 #    none          Do NOT send feedback and status
-method = messagebus
+method = @PIPELINE_FEEDBACK_METHOD@

LCS/MessageBus/include/MessageBus/Message.h

@@ -27,7 +27,7 @@
 #include <qpid/messaging/Message.h>
 #include <qpid/types/Variant.h>
 #else
-#include <MessageBus/QpidMock.h>
+#include <MessageBus/NoQpidFallback.h>
 #endif
 
 #include <string>

LCS/MessageBus/include/MessageBus/MsgBus.h

@@ -31,7 +31,7 @@
 #include <qpid/messaging/Session.h>
 #include <qpid/messaging/Address.h>
 #else
-#include <MessageBus/QpidMock.h>
+#include <MessageBus/NoQpidFallback.h>
 #endif
 
 #include <Common/Exception.h>

LCS/MessageBus/include/MessageBus/QpidMock.h → LCS/MessageBus/include/MessageBus/NoQpidFallback.h

-//# QpidMock.h: A fake implementation of the QPID API
+//# NoQpidFallback.h: A fake implementation of the QPID API in case QPID is not installed
 //#
 //# Copyright (C) 2015
 //# ASTRON (Netherlands Institute for Radio Astronomy)
@@ -20,8 +20,8 @@
 //#
 //# $Id$
 
-#ifndef LOFAR_MESSAGEBUS_QPIDMUCK_H
-#define LOFAR_MESSAGEBUS_QPIDMUCK_H
+#ifndef LOFAR_MESSAGEBUS_NO_QPID_FALLBACK_H
+#define LOFAR_MESSAGEBUS_NO_QPID_FALLBACK_H
 
 #ifndef HAVE_QPID
 #include <Common/LofarLogger.h>

LCS/MessageBus/src/CMakeLists.txt

@@ -8,7 +8,7 @@ set(messagebus_LIB_SRCS
   LogSink.cc
   MsgBus.cc
   Message.cc
-  QpidMock.cc)
+  NoQpidFallback.cc)
 
 set(messagebus_PROGRAMS
 )
@@ -23,6 +23,7 @@ python_install(
   __init__.py
   msgbus.py
   message.py
+  noqpidfallback.py
   DESTINATION lofar/messagebus)
 
 add_subdirectory(protocols)

LCS/MessageBus/src/QpidMock.cc → LCS/MessageBus/src/NoQpidFallback.cc

-//# QpidMock.cc: A fake implementation of the QPID API
+//# NoQpidFallback.cc: A fake implementation of the QPID API in case QPID is not installed
 //#
 //# Copyright (C) 2015
 //# ASTRON (Netherlands Institute for Radio Astronomy)
@@ -23,7 +23,7 @@
 #include <lofar_config.h>
 
 #ifndef HAVE_QPID
-#include <MessageBus/QpidMock.h>
+#include <MessageBus/NoQpidFallback.h>
 
 std::ostream& operator<<(std::ostream& out, const qpid::types::Variant& value) {
   (void)value;

LCS/MessageBus/src/message.py

@@ -17,10 +17,11 @@
 # with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
 
 try:
-  import qpid.messaging
-  enabled = True
+  import qpid.messaging as messaging
+  MESSAGING_ENABLED = True
 except ImportError:
-  enabled = False
+  import noqpidfallback as messaging
+  MESSAGING_ENABLED = False
 
 import xml.dom.minidom as xml
 import xml.parsers.expat as expat
@@ -61,7 +62,7 @@ def _uuid():
   """
     Return an UUID
   """
-  return str(qpid.messaging.uuid4())
+  return str(messaging.uuid4())
 
 class MessageException(Exception):
     pass
@@ -145,7 +146,7 @@ class MessageContent(object):
     def qpidMsg(self):
       """ Construct a NEW QPID message. """
 
-      msg = qpid.messaging.Message(content_type="text/plain", durable=True)
+      msg = messaging.Message(content_type="text/plain", durable=True)
       msg.content = self.content()
 
       return msg

LCS/MessageBus/src/msgbus.py

@@ -19,10 +19,11 @@
 # id.. TDB
 
 try:
-  import qpid.messaging
-  enabled = True
+  import qpid.messaging as messaging
+  MESSAGING_ENABLED = True
 except ImportError:
-  enabled = False
+  import noqpidfallback as messaging
+  MESSAGING_ENABLED = False
 
 import os
 import signal
@@ -47,14 +48,14 @@ class Session:
         self.closed = False
 
         logger.info("[Bus] Connecting to broker %s", broker)
-        self.connection = qpid.messaging.Connection(broker)
+        self.connection = messaging.Connection(broker)
         self.connection.reconnect = True
         logger.info("[Bus] Connected to broker %s", broker)
 
         try:
             self.connection.open()
             self.session = self.connection.session() 
-        except qpid.messaging.MessagingError, m:
+        except messaging.MessagingError, m:
             raise BusException(m)
 
         # NOTE: We cannuot use:
@@ -87,7 +88,7 @@ class Session:
         # to data loss if the stall was legit.
         try:
             self.connection.close(5.0)
-        except qpid.messaging.exceptions.Timeout, t:
+        except messaging.exceptions.Timeout, t:
             logger.error("[Bus] Could not close connection: %s", t)
 
     def __enter__(self):
@@ -110,7 +111,7 @@ class ToBus(Session):
 
         try:
             self.sender = self.session.sender(self.address(queue, options))
-        except qpid.messaging.MessagingError, m:
+        except messaging.MessagingError, m:
             raise BusException(m)
 
     def send(self, msg):
@@ -121,11 +122,11 @@ class ToBus(Session):
               # Send Message or MessageContent object
               self.sender.send(msg.qpidMsg())
             except AttributeError:
-              # Send string or qpid.messaging.Message object
+              # Send string or messaging.Message object
               self.sender.send(msg)
 
             logger.info("[ToBus] Message sent to queue %s", self.queue)
-        except qpid.messaging.SessionError, m:
+        except messaging.SessionError, m:
             raise BusException(m)
 
 class FromBus(Session):
@@ -140,7 +141,7 @@ class FromBus(Session):
 
             # Need capacity >=1 for 'self.session.next_receiver' to function across multiple queues
             receiver.capacity = 1 #32
-        except qpid.messaging.MessagingError, m:
+        except messaging.MessagingError, m:
             raise BusException(m)
 
     def get(self, timeout=None):
@@ -156,7 +157,7 @@ class FromBus(Session):
                     logger.error("[FromBus] Could not retrieve available message on queue %s", receiver.source)
                 else:
                     logger.info("[FromBus] Message received on queue %s", receiver.source)
-        except qpid.messaging.exceptions.Empty, e:
+        except messaging.exceptions.Empty, e:
             return None
 
         if msg is None:

LCS/MessageBus/src/noqpidfallback.py

+#!/usr/bin/python
+
+import sys
+print >>sys.stderr, "QPID support NOT enabled! Will NOT connect to any broker, and messages will be lost!"
+
+def uuid4():
+  return "<uuid>"
+
+
+"""
+  Exceptions.
+"""
+
+class MessagingError(Exception):
+  pass
+
+class SessionError(Exception):
+  pass
+
+class exceptions:
+  class Timeout(Exception):
+    pass
+
+  class Empty(Exception):
+    pass
+
+"""
+  Messages.
+"""
+
+class Message(object):
+  def __init__(self, content_type, durable):
+    self.content = ""
+
+"""
+  Communication.
+"""
+
+class Sender(object):
+  def __init__(self, dest):
+    self.dest = dest
+
+  def send(self, msg):
+    pass
+
+class Receiver(object):
+  def __init__(self, source):
+    self.capacity = 0
+    self.source = source
+
+  def fetch(self):
+    return None
+
+class Session(object):
+  def sender(self, address):
+    return Sender(address)
+
+  def receiver(self, address):
+    return Receiver(address)
+
+  def next_receiver(self, timeout=0):
+    return Receiver("unknown")
+
+  def acknowledge(self, msg):
+    pass
+
+class Connection(object):
+  def __init__(self, broker):
+    self.reconnect = False
+
+  def open(self):
+    pass
+
+  def close(self, timeout=0):
+    pass
+
+  def session(self):
+    return Session()
+

LCS/MessageBus/test/tPyMsgBus.py

 #!/usr/bin/python
-from lofar.messagebus.msgbus import FromBus, ToBus
+#
+# Test the basic functionality of FromBus and ToBus, both
+# to send and to forward messages.
+#
+# Note that without QPID support, the classes are still usable,
+# but messages won't arrive. We consider that case also in this test.
+#
+
+from lofar.messagebus.msgbus import FromBus, ToBus, MESSAGING_ENABLED
 from lofar.messagebus.message import Message, MessageContent
 
 # Send a message (send MessageContent)
@@ -12,7 +20,8 @@ tbus.send(tmsg)
 fbus = FromBus("test")
 fmsg = fbus.get(1)
 
-# Verify the content
+# Verify the content (only if QPID support is enabled)
+if MESSAGING_ENABLED:
   assert fmsg != None
   assert fmsg.content().payload == "foo"
 
@@ -20,6 +29,7 @@ assert fmsg.content().payload == "foo"
 tbus.send(fmsg)
 rmsg = fbus.get(1)
 
-# Verify the content
+# Verify the content (only if QPID support is enabled)
+if MESSAGING_ENABLED:
   assert rmsg != None
   assert rmsg.content().payload == "foo"

LCS/MessageDaemons/src/MessageRouter

@@ -62,10 +62,11 @@ class BusMulticast(threading.Thread):
 
         try:
           content = msg.content()
+          log("INFO","[%s] [%s] Message received" % (self.source, content))
         except message.MessageException, e:
           content = "<unknown>"
+          log("WARN","[%s] Non-compliant message received" % (self.source,))
 
-        log("INFO","[%s] [%s] Message received" % (self.source, content))
 
         outdump.send(msg)
         for outbus in outbusses: