/*
 * Copyright 2005-2006 The Apache Software Foundation.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.servicemix.jbi.messaging;

import org.apache.servicemix.JbiConstants;
import org.apache.servicemix.components.util.ComponentSupport;
import org.apache.servicemix.jbi.ExchangeTimeoutException;
import org.apache.servicemix.jbi.container.JBIContainer;
import org.apache.servicemix.jbi.jaxp.StringSource;
import org.apache.servicemix.tck.ExchangeCompletedListener;

import javax.jbi.messaging.DeliveryChannel;
import javax.jbi.messaging.ExchangeStatus;
import javax.jbi.messaging.Fault;
import javax.jbi.messaging.InOnly;
import javax.jbi.messaging.InOptionalOut;
import javax.jbi.messaging.InOut;
import javax.jbi.messaging.MessageExchangeFactory;
import javax.jbi.messaging.MessagingException;
import javax.jbi.messaging.NormalizedMessage;
import javax.jbi.messaging.RobustInOnly;
import javax.jbi.messaging.MessageExchange.Role;
import javax.xml.namespace.QName;

import junit.framework.TestCase;

public class MEPExchangeTest extends TestCase {

	public static final String PAYLOAD = "<payload/>";
	public static final String RESPONSE = "<response/>";
	
	private JBIContainer container;
	private TestComponent provider;
	private TestComponent consumer;
    private ExchangeCompletedListener listener;
	
	public static class TestComponent extends ComponentSupport {
		public TestComponent(QName service, String endpoint) {
			super(service, endpoint);
		}
		public DeliveryChannel getChannel() throws MessagingException {
			return getContext().getDeliveryChannel();
		}
	}
	
	public void setUp() throws Exception {
		// Create jbi container
		container = new JBIContainer();
        container.setEmbedded(true);
		container.setFlowName("st");
        listener = new ExchangeCompletedListener();
        container.addListener(listener);
		container.init();
		container.start();
		// Create components
		provider = new TestComponent(new QName("provider"), "endpoint");
		consumer = new TestComponent(new QName("consumer"), "endpoint");
		// Register components
		container.activateComponent(provider, "provider");
		container.activateComponent(consumer, "consumer");
	}
	
	public void tearDown() throws Exception {
        if (listener != null) {
            listener.assertExchangeCompleted();
        }
		if (container != null) {
			container.shutDown();
		}
		container = null;
	}
	
	public void testInOnly() throws Exception {
		// Send message exchange
		MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
		InOnly mec = mef.createInOnlyExchange();
		NormalizedMessage m = mec.createMessage();
		m.setContent(new StringSource(PAYLOAD));
		mec.setInMessage(m);
        assertEquals(Role.CONSUMER, mec.getRole());
        try {
            mec.setMessage(null, "in");
            fail("Message is null");
        } catch (Exception e) {
            // ok
        }
        try {
            mec.setMessage(mec.createMessage(), "in");
            fail("Message already set");
        } catch (Exception e) {
            // ok
        }
        try {
            mec.setMessage(mec.createMessage(), "out");
            fail("Out not supported");
        } catch (Exception e) {
            // ok
        }
        try {
            mec.setMessage(mec.createFault(), "fault");
            fail("Fault not supported");
        } catch (Exception e) {
            // ok
        }
		consumer.getChannel().send(mec);
        try {
            mec.setProperty("myprop", "myvalue");
            fail("Component is not owner");
        } catch (Exception e) {
            // ok
        }
		// Provider side
		InOnly mep = (InOnly) provider.getChannel().accept(1000L);
		assertNotNull(mep);
		assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        assertEquals(Role.PROVIDER, mep.getRole());
		mep.setStatus(ExchangeStatus.DONE);
		provider.getChannel().send(mep);
		// Consumer side
		assertSame(mec, consumer.getChannel().accept(1000L));
		assertEquals(ExchangeStatus.DONE, mec.getStatus());
        assertEquals(Role.CONSUMER, mec.getRole());
		// Nothing left
		assertNull(consumer.getChannel().accept(100L)); // receive in
		assertNull(provider.getChannel().accept(100L)); // receive in
	}
    
    public void testInOnlyWithError() throws Exception {
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOnly mec = mef.createInOnlyExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        assertEquals(Role.CONSUMER, mec.getRole());
        consumer.getChannel().send(mec);
        // Provider side
        InOnly mep = (InOnly) provider.getChannel().accept(1000L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        assertEquals(Role.PROVIDER, mep.getRole());
        mep.setError(new Exception());
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(1000L));
        assertEquals(ExchangeStatus.ERROR, mec.getStatus());
        assertEquals(Role.CONSUMER, mec.getRole());
        // Check we can not send the exchange anymore
        try {
            mec.setStatus(ExchangeStatus.DONE);
            consumer.getChannel().send(mec);
            fail("Exchange status is ERROR");
        } catch (Exception e) {
            // ok
        }
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
	
	public void testInOnlySync() throws Exception {
		// Create thread to answer
		new Thread(new Runnable() {
			public void run() {
				try {
					// Provider side
					InOnly mep = (InOnly) provider.getChannel().accept(10000L);
					assertNotNull(mep);
					assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
                    assertEquals(Boolean.TRUE, mep.getProperty(JbiConstants.SEND_SYNC));
					mep.setStatus(ExchangeStatus.DONE);
					provider.getChannel().send(mep);
				} catch (Exception e) {
					e.printStackTrace();
					fail();
				}
			}
		}).start();
		// Send message exchange
		MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
		InOnly mec = mef.createInOnlyExchange();
		NormalizedMessage m = mec.createMessage();
		m.setContent(new StringSource(PAYLOAD));
		mec.setInMessage(m);
		boolean result = consumer.getChannel().sendSync(mec, 10000L);
        assertTrue(result);
		assertEquals(ExchangeStatus.DONE, mec.getStatus());
		// Nothing left
		assertNull(consumer.getChannel().accept(100L)); // receive in
		assertNull(provider.getChannel().accept(100L)); // receive in
	}
	
    public void testInOnlySyncWithTimeoutBeforeAccept() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOnly mec = mef.createInOnlyExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        boolean result = consumer.getChannel().sendSync(mec, 100L);
        assertFalse(result);
        assertEquals(ExchangeStatus.ERROR, mec.getStatus());
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
    public void testInOnlySyncWithTimeoutAfterAccept() throws Exception {
        // Create thread to answer
        Thread t = new Thread(new Runnable() {
            public void run() {
                try {
                    // Provider side
                    InOnly mep = (InOnly) provider.getChannel().accept(10000L);
                    assertNotNull(mep);
                    assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
                    assertEquals(Boolean.TRUE, mep.getProperty(JbiConstants.SEND_SYNC));
                    Thread.sleep(100L);
                    mep.setStatus(ExchangeStatus.DONE);
                    provider.getChannel().send(mep);
                } catch (ExchangeTimeoutException e) {
                    // ok
                } catch (Exception e) {
                    e.printStackTrace();
                    fail();
                }
            }
        });
        t.start();
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOnly mec = mef.createInOnlyExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        boolean result = consumer.getChannel().sendSync(mec, 50L);
        
        assertFalse(result);
        assertEquals(ExchangeStatus.ERROR, mec.getStatus());
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        t.join();
    }
    
	public void testInOut() throws Exception {
		// Send message exchange
		MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
		InOut mec = mef.createInOutExchange();
		NormalizedMessage m = mec.createMessage();
		m.setContent(new StringSource(PAYLOAD));
		mec.setInMessage(m);
		consumer.getChannel().send(mec);
		// Provider side
		InOut mep = (InOut) provider.getChannel().accept(100L);
		assertNotNull(mep);
		assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
		m = mep.createMessage();
		m.setContent(new StringSource(RESPONSE));
		mep.setOutMessage(m);
		provider.getChannel().send(mep);
		// Consumer side
		assertSame(mec, consumer.getChannel().accept(100L));
		assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
		mec.setStatus(ExchangeStatus.DONE);
		consumer.getChannel().send(mec);
		// Provider site
		assertSame(mep, provider.getChannel().accept(100L));
		assertEquals(ExchangeStatus.DONE, mec.getStatus());
		// Nothing left
		assertNull(consumer.getChannel().accept(100L)); // receive in
		assertNull(provider.getChannel().accept(100L)); // receive in
	}
	
	public void testInOutSync() throws Exception {
		// Create thread to answer
		new Thread(new Runnable() {
			public void run() {
				try {
					// Provider side
					InOut mep = (InOut) provider.getChannel().accept(10000L);
					assertNotNull(mep);
					assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
					NormalizedMessage m = mep.createMessage();
					m.setContent(new StringSource(RESPONSE));
					mep.setOutMessage(m);
					provider.getChannel().send(mep);
				} catch (Exception e) {
					e.printStackTrace();
					fail();
				}
			}
		}).start();
		// Send message exchange
		MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
		InOut mec = mef.createInOutExchange();
		NormalizedMessage m = mec.createMessage();
		m.setContent(new StringSource(PAYLOAD));
		mec.setInMessage(m);
		consumer.getChannel().sendSync(mec, 10000L);
		assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
		mec.setStatus(ExchangeStatus.DONE);
		consumer.getChannel().send(mec);
		// Provider site
		assertNotNull(provider.getChannel().accept(100L));
		assertEquals(ExchangeStatus.DONE, mec.getStatus());
		// Nothing left
		assertNull(consumer.getChannel().accept(100L)); // receive in
		assertNull(provider.getChannel().accept(100L)); // receive in
	}
	
	public void testInOutSyncSync() throws Exception {
		// Create thread to answer
		Thread t = new Thread(new Runnable() {
			public void run() {
				try {
					// Provider side
					InOut mep = (InOut) provider.getChannel().accept(10000L);
					assertNotNull(mep);
					assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
					NormalizedMessage m = mep.createMessage();
					m.setContent(new StringSource(RESPONSE));
					mep.setOutMessage(m);
					provider.getChannel().sendSync(mep);
					assertEquals(ExchangeStatus.DONE, mep.getStatus());
				} catch (Exception e) {
					e.printStackTrace();
					fail();
				}
			}
		});
		t.start();
		// Send message exchange
		MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
		InOut mec = mef.createInOutExchange();
		NormalizedMessage m = mec.createMessage();
		m.setContent(new StringSource(PAYLOAD));
		mec.setInMessage(m);
		consumer.getChannel().sendSync(mec, 10000L);
		assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
		mec.setStatus(ExchangeStatus.DONE);
		consumer.getChannel().send(mec);
		// Wait until other thread end
		t.join(100L);
		// Nothing left
		assertNull(consumer.getChannel().accept(100L)); // receive in
		assertNull(provider.getChannel().accept(100L)); // receive in
	}
	
    public void testInOutWithFault() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOut mec = mef.createInOutExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        InOut mep = (InOut) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        Fault f = mep.createFault();
        f.setContent(new StringSource(RESPONSE));
        mep.setFault(f);
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
        assertNotNull(mec.getFault());
        mec.setStatus(ExchangeStatus.DONE);
        consumer.getChannel().send(mec);
        // Provider site
        assertSame(mep, provider.getChannel().accept(100L));
        assertEquals(ExchangeStatus.DONE, mec.getStatus());
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
    public void testInOutWithFaultAndError() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOut mec = mef.createInOutExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        InOut mep = (InOut) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        Fault f = mep.createFault();
        f.setContent(new StringSource(RESPONSE));
        mep.setFault(f);
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
        assertNotNull(mec.getFault());
        mec.setStatus(ExchangeStatus.ERROR);
        consumer.getChannel().send(mec);
        // Provider site
        assertSame(mep, provider.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ERROR, mec.getStatus());
        try {
            consumer.getChannel().send(mec);
            fail("Exchange status is ERROR");
        } catch (Exception e) {
            // ok
        }
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
    public void testInOutWithError1() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOut mec = mef.createInOutExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        InOut mep = (InOut) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        m = mep.createMessage();
        m.setContent(new StringSource(RESPONSE));
        mep.setStatus(ExchangeStatus.ERROR);
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ERROR, mec.getStatus());
        try {
            mec.setStatus(ExchangeStatus.DONE);
            consumer.getChannel().send(mec);
            fail("Exchange status is ERROR");
        } catch (Exception e) {
            // ok
        }
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
    public void testInOutWithError2() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOut mec = mef.createInOutExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        InOut mep = (InOut) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        m = mep.createMessage();
        m.setContent(new StringSource(RESPONSE));
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
        mec.setStatus(ExchangeStatus.ERROR);
        consumer.getChannel().send(mec);
        // Provider site
        assertSame(mep, provider.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ERROR, mec.getStatus());
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
	public void testInOptOutWithRep() throws Exception {
		// Send message exchange
		MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
		InOptionalOut mec = mef.createInOptionalOutExchange();
		NormalizedMessage m = mec.createMessage();
		m.setContent(new StringSource(PAYLOAD));
		mec.setInMessage(m);
		consumer.getChannel().send(mec);
		// Provider side
		InOptionalOut mep = (InOptionalOut) provider.getChannel().accept(100L);
		assertNotNull(mep);
		assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
		m = mep.createMessage();
		m.setContent(new StringSource(RESPONSE));
		mep.setOutMessage(m);
		provider.getChannel().send(mep);
		// Consumer side
		assertSame(mec, consumer.getChannel().accept(100L));
		assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
		mec.setStatus(ExchangeStatus.DONE);
		consumer.getChannel().send(mec);
		// Provider site
		assertSame(mep, provider.getChannel().accept(100L));
		assertEquals(ExchangeStatus.DONE, mec.getStatus());
		// Nothing left
		assertNull(consumer.getChannel().accept(100L)); // receive in
		assertNull(provider.getChannel().accept(100L)); // receive in
	}
	
	public void testInOptOutWithoutRep() throws Exception {
		// Send message exchange
		MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
		InOptionalOut mec = mef.createInOptionalOutExchange();
		NormalizedMessage m = mec.createMessage();
		m.setContent(new StringSource(PAYLOAD));
		mec.setInMessage(m);
		consumer.getChannel().send(mec);
		// Provider side
		InOptionalOut mep = (InOptionalOut) provider.getChannel().accept(100L);
		assertNotNull(mep);
		assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
		mep.setStatus(ExchangeStatus.DONE);
		provider.getChannel().send(mep);
		// Consumer side
		assertSame(mec, consumer.getChannel().accept(100L));
		assertEquals(ExchangeStatus.DONE, mec.getStatus());
		// Nothing left
		assertNull(consumer.getChannel().accept(100L)); // receive in
		assertNull(provider.getChannel().accept(100L)); // receive in
	}
	
    public void testInOptOutWithProviderFault() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOptionalOut mec = mef.createInOptionalOutExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        InOptionalOut mep = (InOptionalOut) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        mep.setFault(mep.createFault());
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
        assertNotNull(mec.getFault());
        mec.setStatus(ExchangeStatus.DONE);
        consumer.getChannel().send(mec);
        // Provider site
        assertSame(mep, provider.getChannel().accept(100L));
        assertEquals(ExchangeStatus.DONE, mec.getStatus());
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
    public void testInOptOutWithProviderError() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOptionalOut mec = mef.createInOptionalOutExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        InOptionalOut mep = (InOptionalOut) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        mep.setStatus(ExchangeStatus.ERROR);
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ERROR, mec.getStatus());
        try {
            mec.setStatus(ExchangeStatus.DONE);
            consumer.getChannel().send(mec);
            fail("Exchange status is ERROR");
        } catch (Exception e) {
            // ok
        }
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
    public void testInOptOutWithRepAndConsumerFault() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOptionalOut mec = mef.createInOptionalOutExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        InOptionalOut mep = (InOptionalOut) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        m = mep.createMessage();
        m.setContent(new StringSource(RESPONSE));
        mep.setOutMessage(m);
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
        mec.setFault(mec.createFault());
        consumer.getChannel().send(mec);
        // Provider site
        assertSame(mep, provider.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        assertNotNull(mep.getFault());
        mep.setStatus(ExchangeStatus.DONE);
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.DONE, mec.getStatus());
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
    public void testInOptOutWithRepAndConsumerError() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOptionalOut mec = mef.createInOptionalOutExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        InOptionalOut mep = (InOptionalOut) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        m = mep.createMessage();
        m.setContent(new StringSource(RESPONSE));
        mep.setOutMessage(m);
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
        mec.setStatus(ExchangeStatus.ERROR);
        consumer.getChannel().send(mec);
        // Provider site
        assertSame(mep, provider.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ERROR, mep.getStatus());
        try {
            mep.setStatus(ExchangeStatus.DONE);
            provider.getChannel().send(mep);
            fail("Exchange status is ERROR");
        } catch (Exception e) {
            // ok
        }
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
    public void testInOptOutWithRepFaultAndError() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        InOptionalOut mec = mef.createInOptionalOutExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        InOptionalOut mep = (InOptionalOut) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        m = mep.createMessage();
        m.setContent(new StringSource(RESPONSE));
        mep.setOutMessage(m);
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
        mec.setFault(mec.createFault());
        consumer.getChannel().send(mec);
        // Provider site
        assertSame(mep, provider.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        assertNotNull(mep.getFault());
        mep.setStatus(ExchangeStatus.ERROR);
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ERROR, mec.getStatus());
        try {
            mec.setStatus(ExchangeStatus.DONE);
            consumer.getChannel().send(mec);
            fail("Exchange status is ERROR");
        } catch (Exception e) {
            // ok
        }
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
	public void testRobustInOnly() throws Exception {
		// Send message exchange
		MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
		RobustInOnly mec = mef.createRobustInOnlyExchange();
		NormalizedMessage m = mec.createMessage();
		m.setContent(new StringSource(PAYLOAD));
		mec.setInMessage(m);
		consumer.getChannel().send(mec);
		// Provider side
		RobustInOnly mep = (RobustInOnly) provider.getChannel().accept(100L);
		assertNotNull(mep);
		assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
		mep.setStatus(ExchangeStatus.DONE);
		provider.getChannel().send(mep);
		// Consumer side
		assertSame(mec, consumer.getChannel().accept(100L));
		assertEquals(ExchangeStatus.DONE, mec.getStatus());
		// Nothing left
		assertNull(consumer.getChannel().accept(100L)); // receive in
		assertNull(provider.getChannel().accept(100L)); // receive in
	}
	
    public void testRobustInOnlyWithFault() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        RobustInOnly mec = mef.createRobustInOnlyExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        RobustInOnly mep = (RobustInOnly) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        mep.setFault(mep.createFault());
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
        assertNotNull(mec.getFault());
        mec.setStatus(ExchangeStatus.DONE);
        provider.getChannel().send(mec);
        // Provider site
        assertSame(mep, provider.getChannel().accept(100L));
        assertEquals(ExchangeStatus.DONE, mep.getStatus());
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
    public void testRobustInOnlyWithError() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        RobustInOnly mec = mef.createRobustInOnlyExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        RobustInOnly mep = (RobustInOnly) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        mep.setStatus(ExchangeStatus.ERROR);
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ERROR, mec.getStatus());
        try {
            mec.setStatus(ExchangeStatus.DONE);
            provider.getChannel().send(mec);
            fail("Exchange status is ERROR");
        } catch (Exception e) {
            // ok
        }
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
    public void testRobustInOnlyWithFaultAndError() throws Exception {
        // Send message exchange
        MessageExchangeFactory mef = consumer.getChannel().createExchangeFactoryForService(new QName("provider"));
        RobustInOnly mec = mef.createRobustInOnlyExchange();
        NormalizedMessage m = mec.createMessage();
        m.setContent(new StringSource(PAYLOAD));
        mec.setInMessage(m);
        consumer.getChannel().send(mec);
        // Provider side
        RobustInOnly mep = (RobustInOnly) provider.getChannel().accept(100L);
        assertNotNull(mep);
        assertEquals(ExchangeStatus.ACTIVE, mep.getStatus());
        mep.setFault(mep.createFault());
        provider.getChannel().send(mep);
        // Consumer side
        assertSame(mec, consumer.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ACTIVE, mec.getStatus());
        assertNotNull(mec.getFault());
        mec.setError(new Exception());
        provider.getChannel().send(mec);
        // Provider site
        assertSame(mep, provider.getChannel().accept(100L));
        assertEquals(ExchangeStatus.ERROR, mep.getStatus());
        try {
            mep.setStatus(ExchangeStatus.DONE);
            provider.getChannel().send(mep);
            fail("Exchange status is ERROR");
        } catch (Exception e) {
            // ok
        }
        // Nothing left
        assertNull(consumer.getChannel().accept(100L)); // receive in
        assertNull(provider.getChannel().accept(100L)); // receive in
    }
    
}
