WEB开发网资讯源码教程软件素材CMS中心工具社区
开发学院软件开发Java 消息中介的实用介绍——第 3 部分：使用中介修改消... 阅读

消息中介的实用介绍——第 3 部分：使用中介修改消息

　2009-10-21 00:00:00　来源：WEB开发网　　　
核心提示：引言如果您读过本系列的前两篇文章，消息中介基础（第 1 部分）和使用中介路由消息（第 2 部分），消息中介的实用介绍——第 3 部分：使用中介修改消息，那么消息中介的基础对您来说已不再陌生，消息中介是 IBM® WebSphere® Application Server 的消息传递功能的新的可编程扩展，

引言

如果您读过本系列的前两篇文章，消息中介基础（第 1 部分）和使用中介路由消息（第 2 部分），那么消息中介的基础对您来说已不再陌生。消息中介是 IBM® WebSphere® Application Server 的消息传递功能的新的可编程扩展，可以用于简化使用消息传递的连接系统、服务、应用程序或组件。这两篇文章，连同 IBM 同事最近发表的其他文章，主要介绍了路由消息和简单的日志记录。在第 3 部分中，我们将展示如何使用中介访问和操纵 JMS 消息的实际内容。

中介编程模型允许使用服务数据对象 (SDO) 接口访问消息正文，它简化了多个层和不同数据源之间的数据交换。SDO 规范已通过 JSR 235 提交到 Java Community Process。（要了解 SDO 为何存在以及如何开发使用 SDO 的简单应用程序，请参阅 Introduction to Service Data Objects。）在本文中，我们将采用实际方法在中介中通过 SDO 阅读和撰写消息内容。

在继续阅读本文前，必须确保您已经了解以前的文章内容。本文假定您已经掌握开发、部署和测试简单中介处理程序的方式。如果您尚未读过上述文章，则现在应该读一读。

准备

要开发和测试本文包含的中介，必须安装以下软件组合之一：

IBM WebSphere Application Server Toolkit Version 6.0 和 WebSphere Application Server V6。

IBM Rational® Application Developer V6，包括集成的 WebSphere Application Server V6 测试环境。

IBM Rational Application Developer V6 和 WebSphere Application Server V6。

IBM Rational Software Architect V6.0，包括集成的 WebSphere Application Server V6 测试环境。

还应该下载和应用所安装产品的最新服务级别。在撰写这篇文章时，WebSphere Application Server V6 Refresh Pack 1 (V6.0.1) 是将要发布的最新修订包。请参阅参考资料，以获取关于 WebSphere Application Server 修订包和如何得到它们的详细信息。可以使用 Rational Product Updater（安装任何 Rational 产品时它都会自动安装）获取 Rational 产品的更新程序。或者，从 IBM Rational Support 网站手动下载修订包。

本文中所显示的屏幕图像来自于 Rational Application Developer V6。如果您使用的是 WebSphere Application Server Toolkit，则仍旧可以使用这里给出的步骤，因为这些步骤对于 Rational Application Developer 和工具包来说都是相同的。您可以从 WebSphere Application Server 安装媒体或映像安装 WebSphere Application Server Toolkit。

使用中介访问消息内容

本系列的第 2 部分阐述了如何和为何使用中介路由消息。中介的其他主要功能是可以访问和修改传递中的消息的内容。例如：

中介可用于在两种不同（甚至不兼容）格式之间转换消息内容。

中介可以利用从数据库获取的数据扩充消息内容。

记录消息内容，供调试或审核时使用。

所有这些用途都依赖于对消息中所包含数据的获取和可能的更改。中介最明显的用途是记录消息内容，以供审核或调试使用；例如，在传输的消息图像的前后均做日志记录。稍后，我们将利用这一功能来开发中介，而现在，我们首先看看如何访问消息内容。

中介处理程序框架允许使用消息的 SDO 表示访问消息。要访问消息 SDO，首先需要通过调用 getSIMessage() 方法从 SIMessageContext 获取 SIMessage。获取对 SIMessage 的引用后，即可使用 SIMessage.getDataGraph() 方法获取 SDO DataGraph。消息 SDO DataObject 包含在 DataGraph 根对象中。

这可能听起来有些复杂，但是如清单 1 所示，只要给定了 SIMessageContext，获取消息的 DataObject SDO 表示就是很容易的事了。


清单 1. 获取消息 SDO DataObject
private　DataObject　getMessageDataObject(SIMessageContext　ctx){　
　　　　DataObject　result　=　null;　
　　　　SIMessage　msg　=　ctx.getSIMessage();　
　　　　try　{　
　　　　　　result　=　msg.getDataGraph().getRootObject();　
　　　　}　catch　(SIDataGraphSchemaNotFoundException　e)　{　
　　　　　　e.printStackTrace();　
　　　　}　catch　(SIMessageException　e)　{　
　　　　　　e.printStackTrace();　
　　　　}　
　　　　return　result;　
}　

通过 DataObject，可以使用 XPATH 语法导航结构。JMS 消息的结构非常简单；消息的有效负载包含在 SDO 消息 DataObject 的 data/value 元素中。其属性的类型和内容取决于发送到中介的 JMS 消息的类型。使用 SIMessage.getFormat() 可以获取 JMS 消息的类型，此操作将返回一个字符串，如下表所示：

JMS 消息类型 SIMessage.getFormat() 值
TextMessageJMS：文本
BytesMessageJMS：字节
ObjectMessageJMS：对象
StreamMessageJMS：流

您将注意到上表中没有 JMS MapMessage。当前版本的中介框架不支持 MapMessage 类型，尝试使用这种类型将导致发生异常。JMS 消息类型格式定义在用于引用的 com.ibm.websphere.sib.SIApiConstants 类中。

获取 DataObject 后，只需调用 getString (data/value) 或 getBytes (data/value) 即可检索文本和字节消息的消息内容。检索对象和流消息稍微复杂一些。

检索对象消息

可以通过从有效负载构造 ByteArrayInputStream 来反序列化对象消息，然后再使用它创建 ObjectInputStream 和调用 readObject() 方法。要成功地反序列化消息，需要确保中介处理程序具有访问适当类的权限。清单 2 说明如何反序列化 JMS ObjectMessages。


清单 2. 反序列化 JMS ObjectMessage 的内容
private　Object　getObject(SIMessageContext　ctx)　{　
　　Object　result　=　null;　
　　byte[]　msgBodyBytes　=　null;　
　　DataObject　msgDataObj　=　getMessageDataObject(ctx);　
　　String　format　=　ctx.getSIMessage().getFormat();　
　　if　(msgDataObj　!=null　&&　msgDataObj.isSet("data/value")　{　
　　　　msgBodyBytes　=　msgDataObj.getBytes("data/value");　
　　}　
　　if　((format.equals(SIApiConstants.JMS_FORMAT_OBJECT))　&&　
　　　(msgBodyBytes　!=　null))　{　
　　　　try　{　
　　　　　　ObjectInputStream　in　=　new　ObjectInputStream(　
　　　　　　　　　　new　ByteArrayInputStream(msgBodyBytes));　
　　　　　　result　=　in.readObject();　
　　　　}　catch　(Throwable　e)　{　
　　　　　　//　Error　de-serialising　object　
　　　　　　e.printStackTrace();　
　　　　}　
　　}　
　　return　result;　
}　

检索流消息

流消息包含一个 Java 基元或字节数组的序列。这表示 data/value 元素将包含一个列表。您可以：

使用 getList("data/value")，它返回一个包含流元素的 java.util.List，或者

使用 get("data/value[n]") 访问列表（在该列表中，n 表示第 n 个元素）中的特定项；SDO 索引从 1 开始。

清单 3 是一个显示如何打印 JMS StreamMessage 的内容的示例。
清单 3. 打印 JMS StreamMessage 的内容
private　void　printStreamMessage(SIMessageContext　ctx)　{　
　　DataObject　msgDataObj　=　getMessageDataObject(ctx);　
　　String　msgFormat　=　ctx.getSIMessage().getFormat();　
　　List　msgBodyStream　=　null;　
　　Object　streamItem;　
　　if　(msgDataObj　!=　null　&&　(msgFormat.equals(SIApiConstants.JMS_FORMAT_STREAM)){　
　　　　if　msgDataObj.isSet("data/value"){　
　　　　　　msgBodyStream　=　msgDataObj.getList("data/value");　
　　　　}　
　　}　
　　if　(msgBodyStream　!=　null)　{　
　　　　ListIterator　iterator　=　msgBodyStream.listIterator();　
　　　　for　(int　i　=　1;　iterator.hasNext();　i++)　{　
　　　　　　streamItem　=　iterator.next();　
　　　　　　System.out.print("Item　index:　"　+　i);　
　　　　　　System.out.print("　type:　");　
　　　　　　if　(streamItem　instanceof　byte[])　{　
　　　　　　　　System.out.print("byte[]　data:　");　
　　　　　　　　System.out.println(new　String(　
　　　　　　　　　(byte[])　streamItem));　
　　　　　　}　else　{　
　　　　　　　　System.out.print(streamItem.getClass().getName());　
　　　　　　　　System.out.print("　value:　");　
　　　　　　　　System.out.println(streamItem.toString());　
　　　　　　}　
　　　　}　
　　}　
}　

从上例中可以看出，对于 JMS 消息来说，阅读消息正文相当简单。Web 服务消息，因起源于 XML 而具有更多复杂结构。这使得编写访问 Web 服务消息的中介也更加复杂。如上所述，SIMessage 包含一个 SDO 数据图形。使用 XPATH，可以从数据图形的根（通过调用 getRootObject() 获取）导航结构。SDO 编程模型也可以提供对元数据的访问；例如，可以使用 getType() 和 getProperties() 来发现任何给定 DataObject 的运行时类型和属性。

LoggingMediation：DebugMediation 回顾

在第 1 部分中，我们介绍了中介的概念，同时提到一个简单的调试中介。想一下该中介如何将关于消息的会话和上下文数据简单地写入日志。根据上一节的内容，我们现在来看看如何重写 DebugMediation 以使其更加可重用，以及如何扩展其功能以包括对消息内容的记录。

我们将把日志编写功能分离到中介处理程序的不同类中，而不再编写实现所需 MediationHandler 接口的 Java 类。这有助于重用，允许其他处理程序在不强制继承 DebugMediation 类的情况下进行记录。

对于这一示例，我们将创建一个 MediationLogWriter 类，来使用特定于应用服务器构造的日志名称和级别为其编写条目。MediationLogWriter 使用 MessageBodyFormatter 类，我们还需要为其编码。最后，我们编码用于测试日志写入器的简单中介。下面记录的步骤假定您已经了解了前面文章的示例。

利用前面文章中使用的工作区启动部署环境。

将 mediation.handlers.MediationLogWriter 类的源代码从下载文件导入 MessageMediation 项目。源代码如清单 4 所示，以供参考。此代码的大部分基于第一篇文章的 DebugMediation。除了某些格式改进和少量重构外，MediationLogWriter 现在调用 MessageBodyFormatter，以便在将 JMS 消息正文写入日志前为其编排格式。

清单 4. MediationLogWriter

package　mediation.handlers;　
import　java.util.Iterator;　
import　java.util.List;　
import　java.util.logging.Level;　
import　java.util.logging.Logger;　
import　com.ibm.websphere.sib.SIMessage;　
import　com.ibm.websphere.sib.mediation.messagecontext.SIMessageContext;　
import　com.ibm.websphere.sib.mediation.session.SIMediationSession;　
public　class　MediationLogWriter　{　
　　private　Logger　logger;　
　　private　Level　level;　
　　private　MessageBodyFormatter　msgBodyFmt　=　new　MessageBodyFormatter();　
　　private　MediationLogWriter()　{　
　　　　super();　
　　}　
　　public　MediationLogWriter(Logger　loggerToUse,　Level　levelToLogAt)　{　
　　　　this();　
　　　　logger　=　loggerToUse;　
　　　　level　=　levelToLogAt;　
　　}　
　　public　void　writeMediationInvoked(SIMessageContext　ctx)　{　
　　　　StringBuffer　buf　=　new　StringBuffer("Mediation　invoked　(");　
　　　　appendCommonHeading(buf,　ctx);　
　　　　buf.append(')');　
　　　　logger.log(level,　buf.toString());　
　　}　
　　public　void　writeAll(SIMessageContext　ctx)　{　
　　　　StringBuffer　buf　=　new　StringBuffer();　
　　　　String　indent　=　"　　";　
　　　　appendCommonHeading(buf,　ctx);　
　　　　appendContext(buf,　ctx,　indent);　
　　　　appendSession(buf,　ctx,　indent);　
　　　　appendMessageHeaders(buf,　ctx,　indent);　
　　　　appendMessageBody(buf,　ctx,　indent);　
　　　　logger.log(level,　buf.toString());　
　　}　
　　public　void　writeSession(SIMessageContext　ctx)　{　
　　　　StringBuffer　buf　=　new　StringBuffer();　
　　　　appendCommonHeading(buf,　ctx);　
　　　　appendSession(buf,　ctx,　"");　
　　　　logger.log(level,　buf.toString());　
　　}　
　　private　void　appendSession(StringBuffer　buf,　SIMessageContext　ctx,　
　　　　　　String　indent)　{　
　　　　String　newLine　=　'\n'　+　indent;　
　　　　SIMediationSession　session　=　ctx.getSession();　
　　　　buf.append("\nMediation　Session:"　+　newLine);　
　　　　buf.append("Mediation　=　"　+　session.getMediationName()　+　newLine);　
　　　　buf.append("Destination　=　"　+　session.getDestinationName()　+　newLine);　
　　　　buf.append("Service　Bus　=　"　+　session.getBusName()　+　newLine);　
　　　　buf.append("Messaging　Engine　=　"　+　session.getMessagingEngineName());　
　　　　if　(session.getDiscriminator()　!=　null　
　　　　　　　　&&　!session.getDiscriminator().equals(""))　{　
　　　　　　buf.append(newLine　+　"Discriminator　=　");　
　　　　　　buf.append(session.getDiscriminator());　
　　　　}　
　　　　if　(session.getMessageSelector()　!=　null　
　　　　　　　　&&　!session.getMessageSelector().equals(""))　{　
　　　　　　buf.append(newLine　+　"Message　selector　=　");　
　　　　　　buf.append(session.getDiscriminator());　
　　　　}　
　　}　
　　public　void　writeContext(SIMessageContext　ctx)　{　
　　　　StringBuffer　buf　=　new　StringBuffer();　
　　　　appendCommonHeading(buf,　ctx);　
　　　　appendContext(buf,　ctx,　"");　
　　　　logger.log(level,　buf.toString());　
　　}　
　　private　void　appendContext(StringBuffer　buf,　SIMessageContext　ctx,　
　　　　　　String　indent)　{　
　　　　Iterator　i　=　ctx.getPropertyNames();　
　　　　String　newLine　=　'\n'　+　indent;　
　　　　buf.append("\nMessage　Context:");　
　　　　if　(i　!=　null　&&　!i.hasNext())　{　
　　　　　　String　propName　=　null;　
　　　　　　Object　propValue;　
　　　　　　for　(;　i.hasNext();)　{　
　　　　　　　　propName　=　(String)　i.next();　
　　　　　　　　if　(propName　!=　null　&&　!propName.equals(""))　{　
　　　　　　　　　　buf.append(newLine　+　indent);　
　　　　　　　　　　buf.append(propName);　
　　　　　　　　　　buf.append("　=　");　
　　　　　　　　　　try　{　
　　　　　　　　　　　　propValue　=　ctx.getProperty(propName);　
　　　　　　　　　　　　buf.append('('　+　propValue.getClass().getName()　+　")　");　
　　　　　　　　　　　　buf.append(propValue.toString());　
　　　　　　　　　　}　catch　(Exception　e)　{　
　　　　　　　　　　　　buf.append("exception　");　
　　　　　　　　　　　　buf.append(e.toString());　
　　　　　　　　　　　　buf.append("getting　value");　
　　　　　　　　　　}　
　　　　　　　　}　
　　　　　　}　
　　　　　　if　(propName　!=　null　&&　!propName.equals(""))　{　
　　　　　　　　buf.append("[]");　
　　　　　　　　buf.append(newLine);　
　　　　　　}　
　　　　}　else　{　
　　　　　　buf.append("[]");　
　　　　　　buf.append(newLine);　
　　　　}　
　　}　
　　public　void　writeMessageHeaders(SIMessageContext　ctx)　{　
　　　　StringBuffer　buf　=　new　StringBuffer();　
　　　　appendCommonHeading(buf,　ctx);　
　　　　appendMessageHeaders(buf,　ctx,　"");　
　　　　logger.log(level,　buf.toString());　
　　}　
　　private　void　appendMessageHeaders(StringBuffer　buf,　SIMessageContext　ctx,　
　　　　　　String　indent)　{　
　　　　String　newLine　=　'\n'　+　indent;　
　　　　SIMessage　msg　=　ctx.getSIMessage();　
　　　　buf.append("Message　Headers:"　+　newLine);　
　　　　buf.append("API　message　id　=　"　+　msg.getApiMessageId()　+　newLine);　
　　　　buf.append("System　message　id　=　"　+　msg.getSystemMessageId()　+　newLine);　
　　　　buf.append("Correlation　id　=　"　+　msg.getCorrelationId()　+　newLine);　
　　　　buf.append("Message　format　=　\""　+　msg.getFormat()　+　'\"'　+　newLine);　
　　　　buf.append("Message　descriminator　=　"　+　msg.getDiscriminator()　
　　　　　　　　+　newLine);　
　　　　List　list　=　msg.getUserPropertyNames();　
　　　　buf.append("User　properties:　");　
　　　　Iterator　i;　
　　　　if　(list　!=　null　&&　!list.isEmpty())　{　
　　　　　　i　=　list.iterator();　
　　　　　　String　propName;　
　　　　　　Object　propValue;　
　　　　　　for　(;　i.hasNext();)　{　
　　　　　　　　buf.append(newLine　+　indent);　
　　　　　　　　propName　=　(String)　i.next();　
　　　　　　　　buf.append(propName);　
　　　　　　　　buf.append("　=　");　
　　　　　　　　try　{　
　　　　　　　　　　propValue　=　msg.getUserProperty(propName);　
　　　　　　　　　　buf.append('('　+　propValue.getClass().getName()　+　")　");　
　　　　　　　　　　buf.append(propValue.toString());　
　　　　　　　　}　catch　(Exception　e)　{　
　　　　　　　　　　buf.append("exception　");　
　　　　　　　　　　buf.append(e.toString());　
　　　　　　　　　　buf.append("getting　value");　
　　　　　　　　}　
　　　　　　}　
　　　　　　buf.append(newLine);　
　　　　}　else　{　
　　　　　　buf.append("[]");　
　　　　　　buf.append(newLine);　
　　　　}　
　　　　buf.append("Forward　routing　path　=　"　+　msg.getForwardRoutingPath()　
　　　　　　　　+　newLine);　
　　　　buf.append("Reverse　routing　path　=　"　+　msg.getReverseRoutingPath()　
　　　　　　　　+　newLine);　
　　　　buf.append("Reliability　=　"　+　msg.getReliability()　+　newLine);　
　　　　buf.append("Priority　=　"　+　msg.getPriority()　+　newLine);　
　　　　buf　
　　　　　　　　.append("Redelivered　Count　=　"　+　msg.getRedeliveredCount()　
　　　　　　　　　　　　+　newLine);　
　　　　buf.append("User　id　=　"　+　msg.getUserId());　
　　}　
　　public　void　writeMessageBody(SIMessageContext　ctx)　{　
　　　　StringBuffer　buf　=　new　StringBuffer();　
　　　　appendCommonHeading(buf,　ctx);　
　　　　appendMessageBody(buf,　ctx,　"");　
　　　　logger.log(level,　buf.toString());　
　　}　
　　public　void　appendCommonHeading(StringBuffer　buf,　SIMessageContext　ctx)　{　
　　　　buf.append("Msg　");　
　　　　buf.append(ctx.getSIMessage().getApiMessageId());　
　　　　buf.append("　Mediation:　");　
　　　　buf.append(ctx.getSession().getMediationName());　
　　　　buf.append("　Destination:　");　
　　　　buf.append(ctx.getSession().getDestinationName());　
　　}　
　　private　void　appendMessageBody(StringBuffer　buf,　SIMessageContext　ctx,　
　　　　　　String　indent)　{　
　　　　msgBodyFmt.formatJMSMessage(buf,　ctx.getSIMessage(),　indent);　
　　}　
}　

在 MediationHandlers 项目中创建一个新类 mediation.handlers.MessageBodyFormatter，并添加所需的导入语句（清单 5）。

清单 5. 导入语句

import　java.io.ByteArrayInputStream;　
import　java.io.ObjectInputStream;　
import　java.util.List;　
import　com.ibm.websphere.sib.SIApiConstants;　
import　com.ibm.websphere.sib.SIMessage;　
import　commonj.sdo.DataObject;　

添加方法 formatJMSMessage(StringBuffer buf, SIMessage msg, String indent)，theMediationLogWriter 将调用它（清单 6）。

清单 6. formatJMSMessage

public　void　formatJMSMessage(StringBuffer　buf,　SIMessage　msg,　String　indent)　{　
　　String　msgfmt　=　msg.getFormat();　
　　DataObject　msgRoot;　
　　String　newLine　=　'\n'　+　indent;　
　　try　{　
　　　　msgRoot　=　msg.getDataGraph().getRootObject();　
　　　　if　(msgfmt.equals(SIApiConstants.JMS_FORMAT_TEXT))　{　
　　　　　　buf.append("\nJMS　TextMessage　:"　+　newLine);　
　　　　　　String　msgText　=　msgRoot.getString("data/value");　
　　　　　　buf.append(msgText.replaceAll("\n",　newLine));　
　　　　　　buf.append("\nJMS　TextMessage,　as　bytes　:");　
　　　　　　byte[]　msgBytes　=　msgRoot.getString("data/value").getBytes();　
　　　　　　appendBytes(buf,　msgBytes,　indent);　
　　　　}　else　if　(msgfmt.equals(SIApiConstants.JMS_FORMAT_BYTES))　{　
　　　　　　buf.append("\nJMS　BytesMessage　:　");　
　　　　　　appendBytes(buf,　msgRoot.getBytes("data/value"),　indent);　
　　　　}　else　if　(msgfmt.equals(SIApiConstants.JMS_FORMAT_OBJECT))　{　
　　　　　　buf.append("\nJMS　ObjectMessage　:　(");　
　　　　　　appendObjectMessage(buf,　indent,　msgRoot,　newLine);　
　　　　}　else　if　(msgfmt.equals(SIApiConstants.JMS_FORMAT_STREAM))　{　
　　　　　　buf.append("\nJMS　StreamMessage　:");　
　　　　　　appendStreamMessage(buf,　indent,　msgRoot,　newLine);　
　　　　}　
　　}　catch　(Exception　e)　{　
　　　　buf.append("Cannot　format　message　("　+　msg.getApiMessageId());　
　　　　buf.append(")　body　using　format　"　+　msgfmt　+　"　due　to　"　+　e);　
　　}　
}　

添加方法 appendStreamMessage(StringBuffer buf, String indent, DataObject msgRoot, String newLine) 为流消息编排格式（清单 7）。

清单 7. appendStreamMessage

private　void　appendStreamMessage(StringBuffer　buf,　String　indent,　
　　　　DataObject　msgRoot,　String　newLine)　{　
　List　streamList　=　msgRoot.getList("data/value");　
　Object　streamItem;　
　if　(streamList　!=　null　&&　!streamList.isEmpty())　{　
　　　for　(int　i　=　1;　i　<=　streamList.size();　i++)　{　
　　　　　streamItem　=　msgRoot.get("data/value["　+　i　+　"]");　
　　　　　buf.append(newLine);　
　　　　　buf.append("Stream　item　"　+　i　+　"　:　(");　
　　　　　if　(streamItem　instanceof　byte[])　{　
　　　　　　　buf.append("byte[])");　
　　　　　　　appendBytes(buf,　(byte[])　streamItem,　"　　"　+　indent);　
　　　　　}　else　{　
　　　　　　　buf.append(streamItem.getClass().getName());　
　　　　　　　buf.append(")　"　+　streamItem.toString());　
　　　　　}　
　　　}　
　}　else　{　
　　　buf.append("[]");　
　}　
}　

添加两个方法 appendObjectMessage(StringBuffer buf, String indent, DataObject msgRoot, String newLine) 和 getObject(DataObject msgRoot, String string) 为 ObjectMessages 编排格式（清单 8）。

清单 8. appendObjectMessage 和 getObject

private　void　appendObjectMessage(StringBuffer　buf,　String　indent,　
　　　　DataObject　msgRoot,　String　newLine)　{　
　Object　obj　=　null;　
　try　{　
　　　obj　=　getObject(msgRoot,　"data/value");　
　　　buf.append(obj.getClass().getName());　
　　　String　objString　=　obj.toString();　
　　　if　(objString.indexOf('\n')>0){　
　　　　　buf.append(")\n");　
　　　　　objString　=　indent　+　objString.replaceAll("\n",　newLine);　
　　　　　buf.append(objString);　
　　　}　else　{　
　　　　　buf.append(")　"　+　objString);　　　
　　　}　
　}　catch　(RuntimeException　e)　{　
　　　buf.append(e.getMessage()　+　"　due　to　"　+　e.getCause());　
　}　
}　
private　Object　getObject(DataObject　msgRoot,　String　string)　{　
　　Object　result　=　null;　
　　byte[]　msgBodyBytes　=　msgRoot.getBytes("data/value");　
　　try　{　
　　　　ObjectInputStream　in　=　new　ObjectInputStream(　
　　　　　　　　new　ByteArrayInputStream(msgBodyBytes));　
　　　　result　=　in.readObject();　
　　}　catch　(Throwable　e)　{　
　　　　//　Error　de-serialising　object　
　　　　throw　new　RuntimeException("Error　deserialising　object",　e);　
　　}　
　　return　result;　
}　

最后，添加方法 appendBytes(StringBuffer buf, byte[] bytes, String indent)，将字节数组的格式编排为熟悉的十六进制数列和相关的 ASCII 字符（清单 9）。

清单 9. appendBytes

private　void　appendBytes(StringBuffer　buf,　byte[]　bytes,　
　　　　String　indent)　{　
　　int　length　=　bytes.length;　
　　String　newLine　=　'\n'　+　indent;　　　　
　　for　(int　lineStart　=　0;　lineStart<length;　lineStart　+=　16)　{　
　　　　int　lineEnd　=　Math.min(lineStart+16,　length);　
　　　　StringBuffer　hex=new　StringBuffer();　
　　　　StringBuffer　ascii=new　StringBuffer();　
　　　　for　(int　i=lineStart;　i<lineEnd;　i++)　{　
　　　　　int　b　=　bytes[i];　
　　　　　b=(b+256)%256;　
　　　　　int　c1=b/16;　
　　　　　int　c2=b%16;　
　　　　　hex.append((char)(c1<10　?　'0'+c1　:　'a'+c1-10));　
　　　　　hex.append((char)(c2<10　?　'0'+c2　:　'a'+c2-10));　
　　　　　if　(i%2　==　1)　hex.append('　');　
　　　　　if　((b>=0x20　&&　b<=0x7e))　ascii.append((char)b);　
　　　　　else　ascii.append('.');　
　　　　}　
　　　　int　pad=16-(lineEnd-lineStart);　
　　　　int　spaces=(pad*5　+　pad%2)/2;　
　　　　spaces+=3;　
　　　　for　(int　i=0;　i<spaces;　i++)　hex.append('　');　　　　
　　　　String　offset="0000"+Integer.toHexString(lineStart);　
　　　　offset=offset.substring(offset.length()-4);　
　　　　buf.append(newLine);　
　　　　buf.append(offset);　
　　　　buf.append("　");　
　　　　buf.append(hex.toString());　
　　　　buf.append(ascii.toString());　
　　　}　
}　

进行下一步之前，保存和编译两个类。

现在编码将使用 MediationLogWriter 的 LoggingMediation 处理程序类。将以下中介添加到 MediationHandlers 项目中（清单 10）。

清单 10. 新中介

package　mediation.handlers;　
import　java.util.logging.Level;　
import　java.util.logging.Logger;　
import　javax.xml.rpc.handler.MessageContext;　
import　com.ibm.websphere.sib.mediation.handler.MediationHandler;　
import　com.ibm.websphere.sib.mediation.handler.MessageContextException;　
import　com.ibm.websphere.sib.mediation.messagecontext.SIMessageContext;　
public　class　LoggingMediation　implements　MediationHandler　{　　
　private　static　final　String　className　=　LoggingMediation.class.getName();　
　private　final　Logger　logger　=　Logger.getLogger(className);　
　private　MediationLogWriter　log;　
　public　boolean　handle(MessageContext　ctx)　throws　MessageContextException　{　
　　　getLog().writeSession((SIMessageContext)　ctx);　
　　　getLog().writeContext((SIMessageContext)　ctx);　
　　　getLog().writeMessageHeaders((SIMessageContext)　ctx);　
　　　getLog().writeMessageBody((SIMessageContext)　ctx);　
　　　//　Alternatively　you　could　log　everything　in　one　go　using　:　
　　　//　getLog().writeAll((SIMessageContext)　ctx);　
　　　//　Or　just　log　a　"Mediation　invoked"　message　using　
　　　//　getLog().writeMediationInvoked((SIMessageContext)ctx);　
　　　return　true;　
　　}　
　private　MediationLogWriter　getLog(){　
　　　if　(log　==　null){　
　　　　　log　=　new　MediationLogWriter(logger,Level.INFO);　
　　　}　
　　　return　log;　
　}　
}　

使用 EJB 部署描述符编辑器的 Mediation Handlers 选项卡将此代码组装到 DeployableMediation 项目中，如图 1 所示。
图 1. 组装的中介处理程序类
消息中介的实用介绍——第 3 部分：使用中介修改消息

将企业应用程序发布到应用服务器，然后运行所提供的 LoggingMediationAdmin.jacl 脚本来配置中介和其他消息传递资源。您应该看到如清单 11 所示的消息。

清单 11. LoggingMediationAdmin 消息

Create　SIB　Queue　
$AdminTask　createSIBDestination　-bus　SimpleBus　-name　loggingQueue　
　　-type　queue　-node　neodogNode　-server　server1　
$AdminTask　createSIBJMSConnectionFactory　server1(cells/....)　-name　CF4　
　　-jndiName　jms/CF4　-busName　SimpleBus　-type　queue　
$AdminTask　createSIBJMSQueue　server1(cells/....)　-name　messageQueue　
　　-jndiName　jms/loggingQueue　-queueName　loggingQueue　
$AdminTask　createSIBMediation　-bus　SimpleBus　-mediationName　
　　LoggingMediation　-handlerListName　LoggingMediation　
$AdminTask　mediateSIBDestination　-bus　SimpleBus　-destinationName　
　　loggingQueue　-mediationName　LoggingMediation　-node　neodogNode　
　　-server　server1　
Configuration　saved　

重新启动应用服务器以更新 JNDI。

使用 Universal Test Client 将消息发送到使用连接 factoryjms/CF4 的目的地 jms/loggingQueue。此操作将生成一组日志条目，如清单 12 所示。

清单 12. 日志条目

[26/05/05　12:26:32:804　BST]　0000007a　LoggingMediat　I　　
　　　Msg　ID:ab6a8a35d4c176a305d19602110a134f0000000000000001　
　　　Mediation:　LoggingMediation　Destination:　loggingQueue　
Mediation　Session:　
Mediation　=　LoggingMediation　
Destination　=　loggingQueue　
Service　Bus　=　SimpleBus　
Messaging　Engine　=　neodogNode.server1-SimpleBus　
[26/05/05　12:26:32:805　BST]　0000007a　LoggingMediat　I　　
　　Msg　ID:ab6a8a35d4c176a305d19602110a134f0000000000000001　
　　Mediation:　LoggingMediation　Destination:　loggingQueue　
Message　Context:　
[26/05/05　12:26:32:811　BST]　0000007a　LoggingMediat　I　　
　　　Msg　ID:ab6a8a35d4c176a305d19602110a134f0000000000000001　
　　　Mediation:　LoggingMediation　Destination:　loggingQueueMessage　Headers:　
API　message　id　=　ID:ab6a8a35d4c176a305d19602110a134f0000000000000001　
System　message　id　=　BAC54242D3B04F6F_9000001　
Correlation　id　=　null　
Message　format　=　"JMS:text"　
Message　descriminator　=　null　
User　properties:　[]　
Forward　routing　path　=　[]　
Reverse　routing　path　=　[]　
Reliability　=　ReliablePersistent　
Priority　=　4　
Redelivered　Count　=　0　
User　id　=　
[26/05/05　12:26:32:815　BST]　0000007a　LoggingMediat　I　　
　　　Msg　ID:ab6a8a35d4c176a305d19602110a134f0000000000000001　
　　　Mediation:　LoggingMediation　Destination:　loggingQueue　
JMS　TextMessage　:　
Sample　text　message　contains　a　
new　line　character　
JMS　TextMessage,　as　bytes　:　
0000:　5361　6d70　6c65　2074　6578　7420　6d65　7373　　Sample　text　mess　
0010:　6167　6520　636f　6e74　6169　6e73　2061　0d0a　　age　contains　a..　
0020:　6e65　7720　6c69　6e65　2063　6861　7261　6374　　new　line　charact　
0030:　6572　　　　　　　　　　　　　　　　　　　　er　

更改消息内容

正如在上面的示例中所看到的，阅读消息正文一点也不复杂，尤其是在仅使用 JMS 字节或文本消息时。您将会高兴地发现，更改消息正文也不再是复杂的事情了。如果回顾一下获取消息正文时，在从 SIMessage 得到的 DataObject 引用上调用 get("data/value")，那么，当您了解到 SIMessage 具有一个对应的 setDataGraph(DataGraph dg, String format) 方法时，就不会感到惊奇了。同样，DataGraph 和 DataObject 类也都具有 setter 方法。下面的代码示例说明了如何设置 JMS 文本消息的内容。


清单 13. 设置 JMS 文本消息的内容
try　{　
　　DataGraph　dg　=　siMsg.getDataGraph();　　　　　　　　　
　　dg.getRootObject().setString("data/value",　msgText);　
　　siMsg.setDataGraph(dg,　siMsg.getFormat();　
　　}　catch　(Exception　e){　
　//　The　actual　exceptions　that　can　be　throw　by　the　setDataGraph　method　are　:　
　　　　　//　SIMessageDomainNotSupportedException,　SIDataGraphSchemaNotFoundException,　
　　　　　//　SIDataGraphFormatMismatchException,　SIMessageException　
　　}　

现在，我们通过编写一个用于修改消息的简单中介处理程序将这一概念付诸实践；我们使用该中介来执行一个简单的文本消息的搜索和替换操作。要替换的文本和实际替换文本均在部署描述符中配置。（如果我们在生产环境中使用类似的中介，则使用上下文属性来配置中介效果会更好一些。）我们的示例还使用了 MediationLogWriter，以在消息的图像前后进行记录。

将中介处理程序类（如清单 14 所示）添加到 MediationHandlers 项目。


清单 14. 中介处理程序类
package　mediation.handlers;　
import　java.util.logging.Level;　
import　java.util.logging.Logger;　
import　javax.xml.rpc.handler.MessageContext;　
import　com.ibm.websphere.sib.SIApiConstants;　
import　com.ibm.websphere.sib.SIMessage;　
import　com.ibm.websphere.sib.mediation.handler.MediationHandler;　
import　com.ibm.websphere.sib.mediation.handler.MessageContextException;　
import　com.ibm.websphere.sib.mediation.messagecontext.SIMessageContext;　
import　commonj.sdo.DataGraph;　
public　class　ReplaceTextMediation　implements　MediationHandler　{　
　　　
　private　final　static　String　className　=　ReplaceTextMediation.class.getName();　
　private　final　Logger　logger　=　Logger.getLogger(className);　
　private　final　MediationLogWriter　log　=　new　MediationLogWriter(logger,　Level.INFO);　
　private　String　textToReplace　=　"Message";　
　private　String　replacementText　=　"MediatedMessage";　
　public　boolean　handle(MessageContext　ctx)　throws　MessageContextException　{　
　　log.writeMediationInvoked((SIMessageContext)ctx);　
　　SIMessage　siMsg　=　((SIMessageContext)ctx).getSIMessage();　
　　if　(siMsg.getFormat().equals(SIApiConstants.JMS_FORMAT_TEXT))　{　
　　　　try　{　
　　　　　DataGraph　dg　=　siMsg.getDataGraph();　　　　　　　　　
　　　　　if　(dg.getRootObject().isSet("data/value")){　
　　　　　　　String　msgText　=　dg.getRootObject().getString("data/value");　
　　　　　　　if　(msgText.indexOf(textToReplace)　>=　0)　{　
　　　　　　　　　log.writeMessageBody((SIMessageContext)ctx);　
　　　　　　　　　msgText　=　msgText.replaceAll(textToReplace,　replacementText);　
　　　　　　　　　dg.getRootObject().setString("data/value",　msgText);　
　　　　　　　　　siMsg.setDataGraph(dg,　siMsg.getFormat());　
　　　　　　　　　log.writeMessageBody((SIMessageContext)ctx);　
　　　　　　　}　
　　　　　}　
　　　　}　catch　(Exception　e)　{　
　　　　　logger.log(Level.SEVERE,　"Could　not　process　message　due　to　exception",　e);　
　　　　}　
　　}else{　
　　　logger.log(Level.INFO,　"Message　{0}　not　changed　becuase　it　is　not　a　text　message",　siMsg.getApiMessageId());　
　　}　
　　return　true;　
　}　
　　public　String　getReplacementText()　{　
　　　　return　replacementText;　
　　}　
　　public　void　setReplacementText(String　replacementText)　{　
　　　　this.replacementText　=　replacementText;　
　　}　
　　public　String　getTextToReplace()　{　
　　　　return　textToReplace;　
　　}　
　　public　void　setTextToReplace(String　textToReplace)　{　
　　　　this.textToReplace　=　textToReplace;　
　　}　
}　

将处理程序类作为名为 ReplaceTextMediation 的新中介处理程序编译和组装到 DeployableMediation 项目中。

公布项目并运行所提供的 jacl 脚本 ReplaceTextMediationAdmin.jacl 来创建中介和所需的消息传递资源。

重新启动服务器。

使用 jms/CF5 将消息发送到 jms/replaceTextQueue，并且在消息正文中包括文本 Message。应用服务器日志将包括与以下内容类似的三个日志条目：


清单 15. 应用服务器日志条目
[27/05/05　12:14:43:165　BST]　00000064　ReplaceTextMe　I　　Mediation　invoked　
　(Msg　ID:82f4131b80304e398e1dfbbb110a134f0000000000000001　
　Mediation:　ReplaceTextMediation　Destination:　sampleMediationQueue)　
[27/05/05　12:14:49:346　BST]　00000064　ReplaceTextMe　I　　
　Msg　ID:82f4131b80304e398e1dfbbb110a134f0000000000000001　
　Mediation:　ReplaceTextMediation　Destination:sampleMediationQueue　
JMS　TextMessage　:　Message　text　
JMS　TextMessage,　as　bytes　:　
0000:　4d65　7373　6167　6520　7465　7874　　　　　　　Message　text　
[27/05/05　12:14:49:388　BST]　00000064　ReplaceTextMe　I　　
　Msg　ID:82f4131b80304e398e1dfbbb110a134f0000000000000001　
　Mediation:　ReplaceTextMediation　Destination:sampleMediationQueue　
JMS　TextMessage　:　MediatedMessage　text　
JMS　TextMessage,　as　bytes　:　
0000:　4d65　6469　6174　6564　4d65　7373　6167　6520　　MediatedMessage　
0010:　7465　7874　　　　　　　　　　　　　　　　　text　

结束语

本系列文章的第 3 部分阐述了如何使用中介阅读和更改消息有效负载。转换正在应用程序间路由的消息是一种强大功能，它允许将两个格式要求不同（甚至 JMS 消息类型要求可能也不同的）的消息的应用程序连接在一起。

简单的 ReplaceTextMediation 示例说明了更改文本消息的唯一方式。如何更改消息正文以及将其更改为什么，具体情况取决于所使用的 JMS 消息的类型和内容。

本文没有谈及如何传递 Web 服务消息。这需要配置 Web Services Gateway 和其他资源。WebSphere Application Server 信息中心包含关于将 Web 服务消息映射到 SDO 消息的信息。请拭目以待，关于 Web 服务消息中介的文章即将发布。

下一期我们将介绍如何使用基于 XSLT 的中介来执行更复杂的消息转换。

本文示例源代码或素材下载

Tags：消息 中介 实用

编辑录入：爽爽　[复制链接] [打 印]
[]
  • 好
  • 好的评价　如果觉得好，就请您
      0%(0)
  • 差
  • 差的评价　如果觉得差，就请您
      0%(0)
更多精彩
赞助商链接

热点阅读
焦点图片
最新推荐
精彩阅读