Page History

package org.apache.servicemix.jbi;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Iterator;
import java.util.Set;
import javax.activation.DataHandler;
import javax.activation.FileDataSource;
import javax.annotation.Resource;
import javax.jbi.messaging.DeliveryChannel;
import javax.jbi.messaging.ExchangeStatus;
import javax.jbi.messaging.MessageExchange;
import javax.jbi.messaging.MessagingException;
import javax.jbi.messaging.NormalizedMessage;
import javax.jbi.messaging.MessageExchange.Role;
import org.apache.log4j.Logger;
import org.apache.servicemix.MessageExchangeListener;
import org.apache.servicemix.jbi.jaxp.StringSource;
import org.apache.servicemix.jbi.util.StreamDataSource;

public class HandlerBean implements MessageExchangeListener
{
    private static final Logger logger = Logger.getLogger(HandlerBean.class);

    @Resource
    private DeliveryChannel channel;

    /*
     * (non-Javadoc)
     * @see org.apache.servicemix.MessageExchangeListener#onMessageExchange(javax.jbi.messaging.MessageExchange)
     */
    public void onMessageExchange(MessageExchange exchange) throws MessagingException
        {
        if (exchange == null)
            {
            return;
            }
        
        // The component acts as a consumer, this means this exchange is received because
        // we sent it to another component. As it is active, this is either an out or a fault
        // If this component does not create / send exchanges, you may just throw an
        // UnsupportedOperationException
        if (exchange.getRole() == Role.CONSUMER)
            {
            onConsumerExchange(exchange);
            }
        
        // The component acts as a provider, this means that another component has requested our
        // service
        // As this exchange is active, this is either an in or a fault (out are send by this
        // component)
        else if (exchange.getRole() == MessageExchange.Role.PROVIDER)
            {
            onProviderExchange(exchange);
            }
        
        // Unknown role
        else
            {
            throw new MessagingException("HandlerBean.onMessageExchange(): Unknown role: " + exchange.getRole());
            }
        }

    /**
     * handles the incoming consumer messages
     * 
     * @param exchange
     * @throws MessagingException
     */
    private void onConsumerExchange(MessageExchange exchange) throws MessagingException 
        {
        // Out message
        if (exchange.getMessage("out") != null)
            {
            exchange.setStatus(ExchangeStatus.DONE);
            channel.send(exchange);
            }
    
        // Fault message
        else if (exchange.getFault() != null)
            {
            exchange.setStatus(ExchangeStatus.DONE);
            channel.send(exchange);
            }
    
        // This is not compliant with the default MEPs
        else
            {
            throw new MessagingException("HandlerBean.onConsumerExchange(): Consumer exchange is ACTIVE, but no out or fault is provided");
            }
        }
    
    /**
     * handles the incoming provider messages
     * 
     * @param exchange
     * @throws MessagingException
     */
    private void onProviderExchange(MessageExchange exchange) throws MessagingException 
        {
        // Exchange is finished
        if (exchange.getStatus() == ExchangeStatus.DONE) 
            {
            return;
            }
        // Exchange has been aborted with an exception
        else if (exchange.getStatus() == ExchangeStatus.ERROR) 
            {
            return;
            }
        // Fault message
        else if (exchange.getFault() != null) 
            {
            exchange.setStatus(ExchangeStatus.DONE);
            channel.send(exchange);
            } 
        else 
            {
            NormalizedMessage in = exchange.getMessage("in");
            
            if (in == null)
                {
                // no in message - strange
                throw new MessagingException("HandlerBean.onProviderExchange(): Exchange has no IN message");
                }
            else
                {
                String fileName = null;
                DataHandler content = null;
                Set attNames = in.getAttachmentNames();
                Iterator it = attNames.iterator();
                
                if (attNames.size()==1)
                    {
                    if (it.hasNext())
                        {
                        try
                            {

                            File file = new File(it.next().toString());
                            fileName = file.getName();
                            content = in.getAttachment(fileName);
   
                            File f = File.createTempFile("tmp_", fileName);
                            FileOutputStream fos = new FileOutputStream(f);
                            content.writeTo(fos);
                            fos.flush();
                            fos.close();
                            
                            // for the sake of simplicity only return the file received as attachment
                            NormalizedMessage out = exchange.createMessage();
        
                            // set the content to dummy xml tag
                            out.setContent(new StringSource("<payload/>"));
                            
                            FileDataSource fds = new FileDataSource(f);
                            InputStream is = fds.getInputStream();
                            
                            // create a handler for the attachment
                            DataHandler dh = new DataHandler(new StreamDataSource(is, fds.getContentType()));
                            
                            // and add it to the message
                            out.addAttachment(fileName, dh);
                            
                            // prepare and send the exchange
                            exchange.setMessage(out, "out");
                            channel.send(exchange);
                            
                            f.deleteOnExit();
                            }
                        catch (IOException ex)
                            {
                            throw new MessagingException("HandlerBean.onProviderExchange(): " + ex.getLocalizedMessage());
                            }
                        }
                    }
                else
                    {
                    // more or less than one attachment
                    throw new MessagingException("HandlerBean.onProviderExchange(): Wrong message format (invalid attachment count)");
                    }
                }
            }
        }
}