An XML Router Example
 | ||
| ||
Now let's develop a simple example program to demonstrate some of the concepts we have been looking at. In this example, we will develop a simple XML router, than receives messages on a queue, examines the content of the message, adds a log entry to the message header, and forwards the resulting message on a queue named by the message. We will use the Point-to-Point messaging domain here, but the idea can be very simply extended to Publish/Subscribe. For example:
First, let's develop the program based on the input message sent to the router, and the desired output messages that the router produces. As input, we will again use our simple stock buy example (keep this in a file called simpleMsg.xml so we can use it later):
<?xml version="1.0" encoding="UTF-8"?> <Envelope> <Header> <Service name="buy"/> </Header> <Body> <Trade> <Stock> Wrox </Stock> <Number> 1000 </Number> </Trade> </Body> </Envelope>
This message will be placed on a queue called routerInput by a simple producer program (the producer we will develop is really just a test scaffolding to introduce messages to the queue). Our message router program synchronously reads from this queue. When a message arrives, it inspects the name attribute of the service element to determine its destination queue. In a real application, there would be some mapping between a local service name and a physical destination queue name. To simplify things for the purposes of the example, we will simply place the output message on a queue named after the service – in this case, "buyService".
It would be useful if our router entered a log entry in the header indicating that it routed the message. This would be useful so that we could track problems if messages traveled over a series of intermediate routers to get to its ultimate destination. The SMTP protocol, for example, does something similar. Every hop that a message takes is recorded in the header. This information is invaluable to mail administrators when diagnosing problems.
So our output message, which the router will place on the buyService queue, will look something like this:
<?xml version="1.0" encoding="UTF-8"?> <Envelope> <Header> <log> Request received by router#1 </log> <Service name="buy"/> </Header> <Body> <Trade> <Stock> Wrox </Stock> <Number> 1000 </Number> </Trade> </Body> </Envelope>
Of course, the router should be able to route to any queue named as a service. We will make the simplifying assumption that the router can create any queue it needs. In a real application of course, queues should probably be fixed by a the administrator of the JMS service provider, and any message containing a service name that doesn't map to a physical queue should be logged as an error.
We will try to make a logical separation of our router into two main parts: a dispatcher, which reads from input queues and writes to output queues; and a message handler, which parses and interprets the input XML message, and produces the output XML message containing the log entry.
We will develop the message handler first. Ideally, the dispatcher should simply send the input text message to the handler, and get an output text message back, along with the destination queue name. This way, the dispatcher is shielded from needing to know anything about the details of XML. We will call our message handler class XmlMsgHandler. Here is the main code defining the class and its attributes. We look at the individual methods, including the constructor in detail as we develop the logic to the code:
package xmlrouter; import java.io.StringWriter; import java.io.StringReader; import java.io.IOException; // JAXP import javax.xml.parsers.DocumentBuilderFactory; import javax.xml.parsers.DocumentBuilder; // SAX import org.xml.sax.InputSource; import org.xml.sax.SAXException; // DOM import org.w3c.dom.Document; import org.w3c.dom.Element; import org.w3c.dom.Node; import org.w3c.dom.NodeList; // TRaX import javax.xml.transform.TransformerFactory; import javax.xml.transform.Transformer; import javax.xml.transform.Templates; import javax.xml.transform.stream.StreamSource; import javax.xml.transform.stream.StreamResult; import javax.xml.transform.dom.DOMSource; public class XmlMsgHandler { private javax.xml.transform.TransformerFactory transformerFactory = null; private javax.xml.transform.Transformer transformer = null; private javax.xml.parsers.DocumentBuilderFactory documentBuilderFactory = null; private javax.xml.parsers.DocumentBuilder documentBuilder = null; private final java.lang.String styleSheetFileName = "msgTransform.xsl"; private javax.xml.transform.Templates templates = null; The imports for JAXP, SAX, and DOM probably look familiar from out earlier examples. The TRaX imports probably will not be familiar.
TRaX is a new addition to the JAXP API to encapsulate XML transformers in a vendor-neutral manner, the same as it does for SAX and DOM. In this case, we will be using TRaX to access the Xerces XSLT processor, which comes with the JAXP distribution. We will not cover XSLT in detail here, but essentially it is a stylesheet language for transforming XML from one form to another. Probably the most common use of XSLT is to transform an XML document containing data into a presentation markup language, like WML or HTML (which XSLT handles as a special case). Here, we will use it simply to add the log entry to the XML message.
It is a trivial example of a very powerful technology, but it does make an interesting point. Once you have a basic dispatcher/handler program written in Java, the actual behavior of the program for different messages can be implemented as a stylesheet. The advantage to this is that additional behavior for new message types can simply be added as new stylesheet that is registered with the application. The Java code never needs to be recompiled. XSLT, although complex, is extremely powerful at transforming XML, and is often much simpler to write than the equivalent Java code.
Furthermore, there are a number of graphic mapping programs becoming available that automatically generate XSLT based on associations created between an input and output DOM. This would allow an administrator with no significant knowledge of XSLT to create transforms between input and output messages. Graphical transform utilities are commonly available for commercial-grade message routers, such as IBM's MQSeries Integrator or Mercator's e-Business Broker.
Let's have a look at the constructor for our message handler:
public XmlMsgHandler() throws ParsingException { //Initialize message handler object transformerFactory = TransformerFactory.newInstance(); try { // Use transformer templates. templates = transformerFactory.newTemplates(new StreamSource(styleSheetFileName)); } catch (javax.xml.transform.TransformerException e) { throw new ParsingException("\nCaught TransformerException: " + e); } documentBuilderFactory = DocumentBuilderFactory.newInstance(); //Turn off validation, and turn off namespaces documentBuilderFactory.setValidating(false); documentBuilderFactory.setNamespaceAware(false); try { documentBuilder = documentBuilderFactory.newDocumentBuilder(); } catch (javax.xml.parsers.ParserConfigurationException e) { throw new ParsingException("Caught ParserConfigurationException: " + e); } } The first thing to notice is that it throws a ParsingException. This is a simple new exception we will define later as a catchall for parsing errors. It simply extends java.lang.Exception.
In the above constructor for XmlMsgHandler, most of the code should be fairly familiar setup from the DOM example we did earlier. Note here we are creating a TRaX template from an XSLT style sheet residing in the file msgTransform.xsl (more on this file later). In a real application, we would load a set of different transforms, each for a different message type. To simplify things, here we will just use one. The template is a compiled version of the stylesheet that we will apply to each incoming message. It acts as a thread-safe factory for transformer objects, which we will encounter later, that actually effect the transform.
Now we need a method so that the handler can pass in a text-based input message, and get back an output message and the queue it should be placed in. We need a simple structure DestinationBoundMsg to contain this output, which we'll look at next. Here is the analysis routine, which returns a DestinationBoundMsg object:
public DestinationBoundMsg analyzeMsg(String textInputMsg) throws ParsingException { // Parse document into DOM tree StringReader stringReader = new StringReader(textInputMsg); InputSource inputSource = new InputSource(stringReader); Document document = null; try { document = documentBuilder.parse(inputSource); document.normalize(); } catch (IOException e) { throw new ParsingException("Caught IOException: " + e); } catch (SAXException e) { throw new ParsingException("Caught SAXException: " + e); } //Get destination name based on query of document String destinationName = getDestinationName(document); //Use XSLT transform to add fields to message and convert into a stream we //can easily convert to a string. StringWriter buffer = new StringWriter(); try { // Get transformer from templates factory. transformer = templates.newTransformer(); transformer.transform(new DOMSource(document), new StreamResult(buffer)); } catch (javax.xml.transform.TransformerException e) { throw new ParsingException("Caught TransformerException: " + e); } return new DestinationBoundMsg(destinationName, buffer.toString()); } As you can see, first we create a DOM document object from the input message. We then query this document for the destination name (we will cover this in a moment). Finally, we take our DOM document and apply a transform to it, adding the log entry. Notice that the transform() method takes an input DOMSource object, and output StreamResult object as parameters. TraX also provides a number of wrappers for different input and output types:
-
DOMSource, DOMResult
-
SAXSource, SAXResult
-
StreamSource, StreamResult
These can be mixed and matched in different pairs, such as SAXSource in and DOMSource out. This provides you with the ability to transform between XML representations. This is very powerful. In the analyzeMsg() method, we are exploiting this to render the output as a stream, which we can easily convert to a string reply message and encapsulate in the output DestinationBoundMsg.
Here is the code to find the service name:
public String getDestinationName(Document document) throws ParsingException {. //Use XPath-like syntax to get service element Element rootElement = document.getDocumentElement(); XmlUtilities xmlUtilities = new XmlUtilities(); NodeListImpl nodeList = xmlUtilities.findElementByQuery( rootElement, "/Header/Service"); if (nodeList.getLength() != 1) { throw new ParsingException("ParsingError: " + "Could not find service or found multiple services."); } Element service = (Element) nodeList.item(0); //Get attribute value from service String serviceName = service.getAttribute("name"); At this point, we would probably use a naming service to map the service attribute to a destination. JNDI would be a good choice, but we could also query a singleton information service that is initialized with an XML properties file. To simplify the example, we will use the serviceName directly, and assume it maps directly to a logical queue or topic name:
return serviceName; } }
Here is the container class for the returned messages:
package xmlrouter; public class DestinationBoundMsg { private String destination = null; private String msg = null; public DestinationBoundMsg() { super(); } public DestinationBoundMsg(String destination, String msg) { this.destination = destination; this.msg = msg; } public java.lang.String getDestination() { return destination; } public java.lang.String getMessage() { return msg; } public java.lang.String getMsg() { return msg; } public void setDestination(java.lang.String newDestination) { destination = newDestination; } public void setMsg(java.lang.String newMsg) { msg = newMsg; } } For completeness, here is the code for the ParsingException:
package xmlrouter; public class ParsingException extends Exception { public ParsingException() { super(); } public ParsingException(String s) { super(s); } } We could simply walk the tree as we saw in the prior DOM examples, but let's try something different. XPath is powerful query syntax for accessing information in XML documents. As we will see later when we examine our stylesheet, it is commonly used in XSLT applications. The simplest form of XPath queries uses a file system like notation to indicate an element. For example, we could isolate the service element in our example using the query: "/Envelope/Header/Service". This simplicity has made XPath a favorite tool among XML developers. XPath actually has very rich query syntax, but we only need to do simple queries like the one above. Rather than use an existing XPath service, we will write our greatly simplified version so that you can become a little more familiar with DOM concepts.
The example above makes a query for the element Service subordinate to Header that in turn is subordinate to the current element (our query is being done from the Envelope element down. Here is the XmlUtilities class. Note that it makes a recursive depth-first search for elements matching the query string. It returns a NodeListImpl, essentially a list of nodes that match the query.
package xmlrouter; import org.w3c.dom.Node; import org.w3c.dom.NodeList; public class XmlUtilities { Return NodeListImpl list of elements matching XPath-like queryString; only supports very simple fully qualified queries like a/b/c:
public NodeListImpl findElementByQuery(Node parent, String queryString) throws ParsingException { NodeListImpl nodeListImpl = new NodeListImpl(); String thisElement; String queryTail; if (queryString.charAt(0) != '/') { throw new ParsingException("Malformed query string: " + queryString); } boolean queryTerminus; int elementEnd = queryString.indexOf("/", 1); if (elementEnd == -1) { queryTerminus = true; thisElement = queryString.substring(1); queryTail = null; } else { queryTerminus = false; thisElement = queryString.substring(1, elementEnd); queryTail = queryString.substring(elementEnd); } //Iterate through list of children. Return all children matching query NodeList nodeList = parent.getChildNodes(); for (int i = 0; i < nodeList.getLength(); i++) { Node child = nodeList.item(i); if (child.getNodeType() == Node.ELEMENT_NODE) { if (child.getNodeName().compareTo(thisElement) == 0) { if (queryTerminus) { nodeListImpl.add(child); } else { nodeListImpl.add(findElementByQuery(child, queryTail)); } } } } return nodeListImpl; } } The NodeListImpl is a concrete implementation of the DOM NodeList interface. Every time we find a node that matches the full query, we place it in the list. As we back up through our recursion stack, we also aggregate multiple lists into a NodeListImpl that is ultimately returned to the caller. Here is the code for the NodeListImpl:
package xmlrouter; import java.util.ArrayList; public class NodeListImpl implements org.w3c.dom.NodeList { private ArrayList nodes = new ArrayList(); public NodeListImpl() { super(); } public void add(NodeListImpl nodeListImpl) { if (nodeListImpl.getLength()>0) { nodes.addAll(nodeListImpl.getNodes()); } } public void add (org.w3c.dom.Node node) { nodes.add(node); } public int getLength() { return nodes.size(); } public ArrayList getNodes() { return nodes; } public org.w3c.dom.Node item(int i) { return (org.w3c.dom.Node) nodes.get(i); } } That's it for the XmlMsgHandler class itself. Here is the stylesheet. This should be called msgTransform.xsl, and should reside in the directory you run the program from:
<?xml version="1.0"?> <xsl:stylesheet xmlns:xsl="http://www.w3.org/1999/XSL/Transform" version="1.0"> <xsl:template><xsl:apply-templates /></xsl:template> <xsl:template name="/"> <xsl:copy> <xsl:apply-templates /> </xsl:copy> </xsl:template> <xsl:template match="Envelope"> <xsl:copy> <xsl:apply-templates /> </xsl:copy> </xsl:template> <xsl:template match="Header"> <xsl:copy> <log>Request received by router#1 </log> <xsl:apply-templates /> </xsl:copy> </xsl:template> <xsl:template match="Body"> <xsl:copy> <xsl:for-each select="/"> <xsl:copy /> </xsl:for-each> <xsl:apply-templates /> </xsl:copy> </xsl:template> <xsl:template match="Service"> <xsl:copy> <xsl:attribute name="name"> <xsl:value-of select="@name" /> </xsl:attribute> </xsl:copy> </xsl:template> </xsl:stylesheet>
We will not try to go into detail about XSLT here. Wrox does have some excellent references, such the XSLT Programmer's Reference ISBN 1-861003-12-9, if you are interested. The thing to notice is the addition of the log entry, under the template match for the header element. Most of the remainder of the style sheet is responsible for echoing existing structure as is.
Now let's consider the JMS dispatcher. In this example, we will make use or the asynchronous message listener model. Our implementation will be single-threaded; however it is easy to extend the example to work with multiple JMS queueSessions in multiple threads. These could potentially be monitoring one input queue, or multiple input queues.
Here is our dispatcher class, called MessageRouter. This is where the main() method resides for this program:
package xmlrouter; import javax.jms.QueueSession; import javax.jms.QueueConnection; import javax.jms.Queue; import javax.jms.JMSException; import java.util.Properties; import javax.jms.QueueConnectionFactory; public class MessageRouter { private Listener listener = null; private QueueConnection queueConnection = null; private QueueSession queueSession = null; public static void main(String[] args) { if (args.length != 1) { System.err.printIn("Usage: java MessageRouter QueueName"); System.exit(0); } print("MessageRouter starting..."); MessageRouter messageRouter = new MessageRouter(); try { messageRouter. init (args); messageRouter.monitor(); } catch (JMSException e) { e.printStackTrace(); } catch (ParsingException e) { e.printStackTrace(); } print("MessageRouter done."); } void init(String[] args) throws JMSException, ParsingException { print("Creating a new connection factory"); JNDIHelper h = new JNDIHelper(); QueueConnectionFactory queueConnectionFactory = (QueueConnectionFactory) h.getConnectionFactory("qcFactory") ; //Create a connection print("Creating a new connection "); queueConnection = queueConnectionFactory.createQueueConnection(); print("Creating a new queue session "); queueSession = queueConnection.createQueueSession(false, QueueSession.CLIENT_ACKNOWLEDGE); queueSession.recover(); //Create a queue with a Queue Name String queueName = args[0]; print("Creating a new queue: " + queueName); Queue queue = queueSession.createQueue(queueName); // Initialize listener object. It's onMessage() method will handle // incoming messages. This could throw a ParsingException if init fails. listener = new Listener(queueSession, queue); //Start connection. print("Starting connection"); queueConnection.start(); } public void monitor() { // Hang around until termination order appears, then cleanup. while (true) { try { Thread.sleep(1000); } catch (Exception e) { //Do nothing } if (listener.terminateFlag) { print("Closing down all resources..."); try { listener.close(); queueConnection.close(); } catch (JMSException jmse) { jmse.printStackTrace(); } break; } } } public static void print (String s) { System.out.printIn("MessageRouter \t:"+s); } } Here is our message listener, called Listener:
package xmlrouter; import javax.jms.QueueSession; import javax.jms.Message; import javax.jms.TextMessage; import javax.jms.Queue; import javax.jms.QueueReceiver; import javax.jms.QueueSender; import javax.jms.JMSException; public class Listener implements javax.jms.MessageListener { private QueueReceiver queueReceiver; private QueueSession queueSession; public boolean terminateFlag = false; private XmlMsgHandler xmlMsgHandler = null; public Listener (QueueSession queueSession, Queue queue) throws JMSException, ParsingException { this.queueSession = queueSession; queueReceiver = queueSession.createReceiver(queue); queueReceiver.setMessageListener(this); xmlMsgHandler = new XmlMsgHandler(); } public void close() throws JMSException { //Have to cleanup outside the onMessage() call or things blow up... print("Starting cleanup"); queueReceiver.setMessageListener(null); queueSession.close(); } public static void print(String s) { System.out.printIn("Listener \t:"+s); } public String getInputMsg(Message message) throws javax.jms.JMSException { TextMessage textMessage = (TextMessage) message; String textInputMsg = textMessage.getText(); textMessage.acknowledge(); return textInputMsg; } public void onMessage (Message message) { try { // Get text contents of message String textInputMsg = getInputMsg(message); print ("Received: "+textInputMsg); // Parse and transform message DestinationBoundMsg destinationBoundMsg = xmlMsgHandler.analyzeMsg(textInputMsg); print("Output Message Destination: " + destinationBoundMsg.getDestination() + "\n"); print("Output Message:\n" + destinationBoundMsg.getMsg()); // Send message on its way. sendMsg(destinationBoundMsg); } catch (ParsingException e) { e.printStackTrace(); } catch (JMSException e) { e.printStackTrace(); } //Program will process until terminateFlag is set to true. //This could be done based on some condition noted in a message. //if (someCondition) // terminateFlag=true; } public void sendMsg(DestinationBoundMsg destinationBoundMsg) throws javax.jms.JMSException { //Send message to indicated destination String queueName = destinationBoundMsg.getDestination(); Queue queue = queueSession.createQueue(queueName); QueueSender queueSender = queueSession.createSender(queue); TextMessage textMessage = queueSession.createTextMessage(); textMessage.setText(destinationBoundMsg.getMsg()); queueSender.send(textMessage); queueSender.close(); } } Note we have commented out the termination check. You could implement some check based on message contents if desired.
Now all we need is a simple producer and consumer. Here is the producer:
package xmlrouter; import javax.jms.QueueConnectionFactory; import javax.jms.QueueConnection; import javax.jms.QueueSession; import javax.jms.QueueSender; import javax.jms.Queue; import javax.jms.TextMessage; import javax.jms.JMSException; public class Producer { public static void main(String[] args) { if (args.length != 2) { System.err.printIn("Usage: java Producer QueueName MsgFileName"); System.exit(0); } //This is provider specific print("Creating a new connection factory"); JNDIHelper h = new JNDIHelper(); QueueConnectionFactory queueConnectionFactory = (QueueConnectionFactory) h.getConnectionFactory("qcFactory"); try { print("Creating a new connection"); QueueConnection queueConnection = queueConnectionFactory.createQueueConnection(); print("Creating a new queue session"); QueueSession queueSession = queueConnection.createQueueSession(false, QueueSession.CLIENT_ACKNOWLEDGE); queueSession.recover(); String queueName = args[0]; print("Creating queue name: " + queueName); Queue queue = queueSession.createQueue(queueName); print("Starting connection"); queueConnection.start(); String fileName = args[1]; print("Reading message from file: " + fileName); String textInputMsg = FileToString.getDocument(fileName); TextMessage textMessage = queueSession.createTextMessage(); textMessage.setText(textInputMsg); print("Sending message... "); QueueSender queueSender = queueSession.createSender (queue); queueSender.send(textMessage); print("Starting cleanup"); queueSender.close(); queueSession.close(); queueConnection.close(); print("Done."); } catch (JMSException e) { e.printStackTrace(); } } public static void print (String s) { System.out.printIn("PRODUCER :"+s); } } The producer reads a message from a file as a string using the FileToString utility class. The program then writes this string to the named queue and exits.
Here is the FileToString utility class:
package xmlrouter; import java.io.*; public class FileToString { public FileToString() { super(); } public static String getDocument(String fileName) { File file = new File(fileName); StringBuffer fileText = new StringBuffer(); try { FileReader fileReader = new FileReader(file); BufferedReader bufferedReader = new BufferedReader(fileReader); String line; while ((line = bufferedReader.readLine()) != null) { fileText.append(line); fileText.append("\n"); } } catch (FileNotFoundException e) { System.err.printIn("Could not find file " + fileName); System.exit(0); } catch (IOException e) { System.err.printIn("IO Exception on file " + fileName); System.exit(0); } return fileText.toString(); } } Here is the consumer:
package xmlrouter; import javax.jms.QueueConnectionFactory; import javax.jms.QueueConnection; import javax.jms.QueueSession; import javax.jms.QueueReceiver; import javax.jms.Queue; import javax.jms.TextMessage; import javax.jms.JMSException; public class Consumer { public static void main(String[] args) { if (args.length != 1) { System.err.printIn("Usage: java Consumer QueueName"); System.exit(0); } //Provider specific for Sun's JMQ print("Creating a new connection factory"); JNDIHelper h = new JNDIHelper(); QueueConnectionFactory queueConnectionFactory = (QueueConnectionFactory) h.getConnectionFactory("qcFactory"); try { print("Creating a new connection "); QueueConnection queueConnection = queueConnectionFactory.createQueueConnection(); print("Creating a new queue session "); QueueSession queueSession = queueConnection.createQueueSession(false, QueueSession.CLIENT_ACKNOWLEDGE); queueSession.recover(); String queueName = args[0]; print("Creating a new queue: " + queueName); Queue queue = queueSession.createQueue(queueName); print("Starting connection"); queueConnection.start(); QueueReceiver queueReceiver = queueSession.createReceiver(queue); boolean forever = true; while (forever) { TextMessage textMessage = (TextMessage) queueReceiver.receive(); String text = textMessage.getText(); print("Consumer received: " + text); textMessage.acknowledge(); } // While print("Starting cleanup"); queueReceiver.close(); queueSession.close(); queueConnection.close(); print("Done."); } catch (JMSException jmse) { jmse.printStackTrace(); } } public static void print(String s) { System.out.printIn("Consumer \t:"+s); } } The consumer runs forever, echoing what it received to standard output. A last class is a helper class called JNDIHelper. It enables all the programs to access a queue connection factory through JNDI. Here, we've chosen to set up the objects in Fiorano, but you could in theory choose any provider:
package xmlrouter; import javax.naming.Context; import javax.naming.InitialContext; import javax.naming.NamingException; import java.util.Properties; import javax.jms.ConnectionFactory; public class JNDIHelper { public static final String ICF="fiorano.jms.rtl.FioranoInitialContextFactory"; public static final String URL="localhost:2001"; public static ConnectionFactory getConnectionFactory(String factory) { ConnectionFactory cf = null; try { Properties p = new Properties(); p.put (Context. INITIAL_CONTEXT_FACTORY, ICF); p.put(Context.PROVIDER_URL, URL); Context ctx = new InitialContext(p); cf = (ConnectionFactory) ctx.lookup(factory); } catch (NamingException n) { n.printStackTrace(); } return cf; } } To run this example, compile all the code in a single directory xmlrouter, ensuring that you have all of the necessary JAR files in your path. You need jms.jar, jndi.jar, and Fiorano's fmprtl.zip for the messaging; jaxp.jar, crimson.jar, and xalan.jar from the JAXP distribution.
We must also set up Administered objects. Set up two queues in the Fiorano administration tool, one called routerInputQueue and one called buy:
Also, set up a queue connection factory called qcFactory:
Note that we used Fiorano extensively in 5, so refer to these chapters if you have any problems. Make certain msgTransform.xsl and simpleMsg.xml are in the current directory.
Then, start the router, the consumer, and finally the producer in different windows. The following script would run the system:
start java xmlrouter.MessageRouter routerInputQueue start java xmlrouter.Consumer buy java xmlrouter.Producer routerInputQueue simpleMsg.xml
For example, in reverse order of our starting them up, we see the following outputs. The producer reads the file and sends the message:
The router receives the message and passes it on:
The consumer picks up the message:
Note that queues are created dynamically here (however in Fiorano, it is necessary to register the queue objects), so you can vary the service names in the message. You may want to start multiple consumers to receive these. The router will print the input message, and the output destination queue and message. The consumer will also print each message it reads.
| ||
EAN: 2147483647
Pages: 154