mytransformer.java

基于eclipse环境下开发的servicemix例程

package tranformer;

import java.io.IOException;
import java.util.Map;

import javax.jbi.messaging.ExchangeStatus;
import javax.jbi.messaging.InOut;
import javax.jbi.messaging.MessageExchange;
import javax.jbi.messaging.MessagingException;
import javax.jbi.messaging.NormalizedMessage;
import javax.jbi.messaging.MessageExchange.Role;
import javax.xml.namespace.QName;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.transform.TransformerException;
import javax.xml.transform.dom.DOMSource;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.servicemix.MessageExchangeListener;
import org.apache.servicemix.components.util.ComponentSupport;
import org.apache.servicemix.jbi.jaxp.SourceTransformer;
import org.apache.servicemix.jbi.util.DOMUtil;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.traversal.NodeIterator;
import org.xml.sax.SAXException;

import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap;

import com.sun.org.apache.xpath.internal.CachedXPathAPI;


public class MyTransformer extends ComponentSupport implements MessageExchangeListener {
    
    private static final Log log = LogFactory.getLog(MyTransformer.class); 

    public MyTransformer() {
    	//定义服务的名称，以便JBI通过此名字来找到此服务组件。
        super(new QName("http://servicemix.org/demo", "mytransformer"), "input");
    }
    
    /**
     * 
     下面的方法不停地来接收信息。
      在JBI里面，所有的信息都是通过MessageExchange传过来，每一个传过来的MessageExcahnge都有一个唯一的确认号ExchangeId。
      一个具体的MessageExchange的内容如下：
		InOut[
		  id: ID:OEM-MICRO-2966-1186566076648-2:0
		  status: Active
		  role: provider
		  service: {http://servicemix.org/demo}receiver
		  endpoint: receiver
		  in: <?xml version="1.0" encoding="UTF-8"?><getHello xmlns="hello">
		         <in0 xmlns="hello">Aihu</in0>
		      </getHello>
		]

     */
    public void onMessageExchange(MessageExchange exchange) throws MessagingException {
    	
    	//下面的方法可以得到exchangeId号
    	String exchangid = exchange.getExchangeId();
    	
    	System.out.println("MessageExchange>>>>"+ exchange);
    	
    	if (exchange.getStatus() == ExchangeStatus.ACTIVE) {	
            
    		//下面的方法得到端口号的名称，通过下面的方法可以得到端口号的名称。
	    	String endPoint = exchange.getEndpoint().getEndpointName();

	    	
	    	if(endPoint.equalsIgnoreCase("mytransformer"))
	    	{
	    		//将MessageExchange转化为InOut的形式。
	    		InOut inout = (InOut) exchange;
                
	    		// 取出NormalizedMessage
	    		NormalizedMessage inMessage = inout.getInMessage();
	    		String dest = getProcessedMessage(exchange);

	    		
	    		QName qName = null;
	    		
	    		if(dest.equalsIgnoreCase("axis"))
	        	    qName = new QName("http://servicemix.org/demo", "axisHello");
	        	else
	        	    qName = new QName("http://servicemix.org/demo", "xfireHello");
	        	
	        	InOut inout1 = createInOutExchange(qName, null, null);
	        	
	        	/*
                  因为是异步方式的调用，所有的请求和相应都是通过JBI来控制，所以这里首先把exchangeId存到
                MessageExchange的property里面，以便以后JBI能够取出来，送回相应的相应。*/
	        	
	        	
	        	inout1.setProperty("requestId", exchangid);
	        	
	        	inout1.setInMessage(inMessage);
	        	
	        	send(inout1);
	    	} else
	    	{
	    		InOut inout = (InOut) exchange;
	    		NormalizedMessage outMessage = inout.getOutMessage();
	    		
	    		/*
	    		取出原始的发出请求的MessageExchange的ExchangeId，通过这个Id来找到原始的MessageExchange
	    		（前面已经将它存入到了ConcurrentHashMap里面，因为每一个MessageExchange都有一个端口Endpoint来监听消息，
	    		这样的话，它就可以从JBI收到消息后送回给服务请求者。*/
	    		
	    		String id = (String) getProperty(inout, "requestId");
	    		
	    		InOut inout33 = (InOut) ht.get(id);

	    		inout33.setOutMessage(outMessage);
	    		
	            send(inout33);

	    	}
	    	
	        if (exchange.getRole() == Role.PROVIDER) {
	           ht.put(exchangid,exchange);
	        } else
	        {
	    	 done(exchange);
	    	
	        }
    	}

    }


    protected Object getProperty(MessageExchange me, String name) {
        return me.getProperty(name);
    }
    
    protected Object getInProperty(MessageExchange me, String name) {
        return me.getMessage("in").getProperty(name);
    }
    
    protected Object getOutProperty(MessageExchange me, String name) {
        return me.getMessage("out").getProperty(name);
    }
    
  
     private Map ht = new ConcurrentHashMap();
     
    protected String getProcessedMessage(MessageExchange me)
    {
    	String destination = null;
    	
    	NormalizedMessage inMessage = me.getMessage("in");
    	
    	if(inMessage != null)
    	{
	    try {
			destination = (String) processNode(inMessage);
		} catch (TransformerException e) {
			e.printStackTrace();
		} catch (ParserConfigurationException e) {
			e.printStackTrace();
		} catch (SAXException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
        }
    	
    	NormalizedMessage outMessage = me.getMessage("out");
    	
    	
    	if(outMessage != null)
    	{
           try {
			destination = (String)  processNode(outMessage);
		} catch (TransformerException e) {
			e.printStackTrace();
		} catch (ParserConfigurationException e) {
			e.printStackTrace();
		} catch (SAXException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
        }
    	
        return destination;
    }

    
    public String processNode(NormalizedMessage nm) throws TransformerException, ParserConfigurationException, SAXException, IOException
    {
    	
 	
    	String destination = null;
    	Document document = null;
    		
    	try {
			document=	(Document) new SourceTransformer().toDOMNode(nm);
		} catch (MessagingException e) {
			e.printStackTrace();
		} catch (TransformerException e) {
			e.printStackTrace();
		} catch (ParserConfigurationException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (SAXException e) {
			e.printStackTrace();
		}
    	
    	String nodeValue = textValueOfXPath(document, "//*[local-name()='in0']");
    	Node node = nodeOfXPath(document, "//*[local-name()='in0']");

	    
        if(nodeValue.indexOf("(xfire)")>-1)
        {
           String modifiedValue = nodeValue.substring(0,nodeValue.indexOf("(xfire)")-1);
           node.getFirstChild().setNodeValue(modifiedValue);  
           destination = "xfire";    
        } else if(nodeValue.indexOf("(axis)")>-1)
	{
           String modifiedValue = nodeValue.substring(0,nodeValue.indexOf("(axis)")-1);
           node.getFirstChild().setNodeValue(modifiedValue);  
           destination = "axis";   
	}
        
        DOMSource domsource = new DOMSource( document );
        
        try {
			nm.setContent(domsource);
		} catch (MessagingException e) {
			e.printStackTrace();
		}
        return destination;
     }
    
    

    protected String textValueOfXPath(Node node, String xpath) throws TransformerException {
        CachedXPathAPI cachedXPathAPI = new CachedXPathAPI();
        NodeIterator iterator = cachedXPathAPI.selectNodeIterator(node, xpath);
        Node root = iterator.nextNode();
        if (root instanceof Element) {
            Element element = (Element) root;
            if (element == null) {
                return "";
            }
            String text = DOMUtil.getElementText(element);
            return text;
        }
        else if (root != null) {
            return root.getNodeValue();
        } else {
            return null;
        }
    }
    
    protected Node nodeOfXPath(Node node, String xpath) throws TransformerException {
        CachedXPathAPI cachedXPathAPI = new CachedXPathAPI();
        NodeIterator iterator = cachedXPathAPI.selectNodeIterator(node, xpath);
        Node root = iterator.nextNode();
        if (root instanceof Element) {
            Element element = (Element) root;
            if (element == null) {
                return null;
            }
            return element;
        }
        else if (root != null) {
            return root;
        } else {
            return null;
        }
    }  
    
    
    
}