import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
/** {@inheritDoc}*/
@Override
protected Conduit getInbuiltBackChannel(Message inMessage)
{
   if (inMessage.getExchange().isOneWay())
   {
      return null;
   }
   final UDPConnectionInfo info = inMessage.get(UDPConnectionInfo.class);
   return new AbstractBackChannelConduit()
   {
      public void prepare(Message message) throws IOException
      {
         message.setContent(OutputStream.class, info.out);
      }
   };
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
public static void closeConduit(Exchange exchange) throws IOException {
    ConduitSelector conduitSelector = null;
    synchronized (exchange) {
        conduitSelector = exchange.get(ConduitSelector.class);
        if (conduitSelector != null) {
            exchange.remove(ConduitSelector.class.getName());
        }
    }

    Conduit selectedConduit = null;
    Message message = exchange.getInMessage() == null ? exchange
            .getOutMessage() : exchange.getInMessage();

    if (conduitSelector != null && message != null) {
        selectedConduit = conduitSelector.selectConduit(message);
        selectedConduit.close(message);
    }

    //TODO the line below was removed, check the impact on the protobuffer importer/exporter
    //selectedConduit.close(message);
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
private Conduit getConduit(Message inMessage) throws IOException {
    Exchange exchange = inMessage.getExchange();
    EndpointReferenceType target = exchange.get(EndpointReferenceType.class);
    Conduit conduit =
        exchange.getDestination().getBackChannel(inMessage, null, target);
    exchange.setConduit(conduit);
    return conduit;
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private MessageObserver createMessageObserver(final Conduit c)
{
   return new MessageObserver() 
   {
       public void onMessage(Message inMessage) 
       {
          LoadingByteArrayOutputStream bout = new LoadingByteArrayOutputStream();
          try 
          {
             IOUtils.copy(inMessage.getContent(InputStream.class), bout);
             inMessage.getExchange().put(InputStream.class, bout.createInputStream());
                   
             Map<String, List<String>> inHeaders = 
                (Map<String, List<String>>)inMessage.get(Message.PROTOCOL_HEADERS);
                   
             inMessage.getExchange().put(Message.PROTOCOL_HEADERS, inHeaders);
             c.close(inMessage);
          } 
          catch (IOException e) 
          {
             //ignore
             Logger.getLogger(SOAPConnectionImpl.class).trace(e);
          }
       }
   };
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
public Conduit getConduit(EndpointInfo ei, EndpointReferenceType target, Bus bus) throws IOException {
    LOG.log(Level.FINE, "Creating conduit for {0}", ei.getAddress());
    if (target == null) {
        target = createReference(ei);
    }
    return new UDPConduit(target, bus);
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
@Override
public void initialize(Client client, Bus bus) {
    checkZmqConfig();
    Conduit conduit = client.getConduit();
    if (!(conduit instanceof ZMQConduit)) {
        throw new ConfigurationException(new Message("ZMQCONFIGFEATURE_ONLY_ZMQ", LOG));
    }
    Endpoint ep = client.getEndpoint();
    changeTransportUriToZmq(ep);
    ZMQConduit zmqConduit = (ZMQConduit)conduit;
    zmqConduit.setZmqConfig(zmqConfig);
    super.initialize(client, bus);
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
/**
 * Called when a Conduit is actually required.
 * 
 * @param message
 * @return the Conduit to use for mediation of the message
 */

public synchronized Conduit selectConduit(Message message) {
	Conduit c = message.get(Conduit.class);
	if (c == null) {

		Exchange exchange = message.getExchange();
		InvocationKey key = new InvocationKey(exchange);
		InvocationContext invocation = getInvocation(key);

		if ((invocation != null) && !invocation.getContext().containsKey(IS_SELECTED)) {
			Endpoint target = getAvailableTarget();
			if (target != null && targetChanged(message, target)) {
				setEndpoint(target);
				message.put(Message.ENDPOINT_ADDRESS, target.getEndpointInfo().getAddress());
				overrideAddressProperty(invocation.getContext());
				invocation.getContext().put(IS_SELECTED, "");
			} else if (target == null) {
				throw new Fault(new IOException("No available targets"));
			}
		}

		message.put(CONDUIT_COMPARE_FULL_URL, Boolean.TRUE);
		c = getSelectedConduit(message);
	}
	if (receiveTimeout != null) {
		HTTPClientPolicy httpClientPolicy = ((HTTPConduit) c).getClient();
		httpClientPolicy.setReceiveTimeout(receiveTimeout);
	}

	return c;
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
/**
 * @param inMessage the incoming message
 * @return the inbuilt backchannel
 */
protected Conduit getInbuiltBackChannel(Message inMessage) {
    //we can pass the message back by looking up the camelExchange from inMessage
    return new BackChannelConduit(inMessage);
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
public Conduit getConduit(EndpointInfo targetInfo) throws IOException {
    return getConduit(targetInfo, null, bus);
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
@Test
public void testRoundTripDestination() throws Exception {

    inMessage = null;
    EndpointInfo conduitEpInfo = new EndpointInfo();
    conduitEpInfo.setAddress("camel://direct:Producer");
    // set up the conduit send to be true
    CamelConduit conduit = setupCamelConduit(conduitEpInfo, true, false);
    final Message outMessage = new MessageImpl();

    endpointInfo.setAddress("camel://direct:EndpointA");
    final CamelDestination destination = setupCamelDestination(endpointInfo, true);

    // set up MessageObserver for handling the conduit message
    MessageObserver observer = new MessageObserver() {
        public void onMessage(Message m) {
            try {
                Exchange exchange = new ExchangeImpl();
                exchange.setInMessage(m);
                m.setExchange(exchange);
                verifyReceivedMessage(m, "HelloWorld");
                //verifyHeaders(m, outMessage);
                // setup the message for
                Conduit backConduit;
                backConduit = getBackChannel(destination, m);
                // wait for the message to be got from the conduit
                Message replyMessage = new MessageImpl();
                sendoutMessage(backConduit, replyMessage, true, "HelloWorld Response");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    };
    MockEndpoint error = context.getEndpoint("mock:error", MockEndpoint.class);
    error.expectedMessageCount(0);
    //this call will active the camelDestination
    destination.setMessageObserver(observer);
    // set is one way false for get response from destination
    // need to use another thread to send the request message
    sendoutMessage(conduit, outMessage, false, "HelloWorld");
    // wait for the message to be got from the destination,
    // create the thread to handler the Destination incoming message

    verifyReceivedMessage(inMessage, "HelloWorld Response");
    error.assertIsSatisfied();
    destination.shutdown();
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
@Test
public void testRoundTripDestinationWithFault() throws Exception {

    inMessage = null;
    EndpointInfo conduitEpInfo = new EndpointInfo();
    conduitEpInfo.setAddress("camel://direct:Producer");
    // set up the conduit send to be true
    CamelConduit conduit = setupCamelConduit(conduitEpInfo, true, false);
    final Message outMessage = new MessageImpl();

    endpointInfo.setAddress("camel://direct:EndpointA");
    final CamelDestination destination = setupCamelDestination(endpointInfo, true);
    destination.setCheckException(true);

    // set up MessageObserver for handling the conduit message
    MessageObserver observer = new MessageObserver() {
        public void onMessage(Message m) {
            try {
                Exchange exchange = new ExchangeImpl();
                exchange.setInMessage(m);
                m.setExchange(exchange);
                verifyReceivedMessage(m, "HelloWorld");
                //verifyHeaders(m, outMessage);
                // setup the message for
                Conduit backConduit;
                backConduit = getBackChannel(destination, m);
                // wait for the message to be got from the conduit
                Message replyMessage = new MessageImpl();
                replyMessage.setContent(Exception.class, new RuntimeCamelException());
                sendoutMessage(backConduit, replyMessage, true, "HelloWorld Fault");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
    };
    
    MockEndpoint error = context.getEndpoint("mock:error", MockEndpoint.class);
    error.expectedMessageCount(1);
    
    //this call will active the camelDestination
    destination.setMessageObserver(observer);
    // set is one way false for get response from destination
    // need to use another thread to send the request message
    sendoutMessage(conduit, outMessage, false, "HelloWorld");
    // wait for the message to be got from the destination,
    // create the thread to handler the Destination incoming message

    verifyReceivedMessage(inMessage, "HelloWorld Fault");
    error.assertIsSatisfied();
    
    destination.shutdown();
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
private Conduit getBackChannel(CamelDestination destination, Message m) throws IOException {
    return destination.getInbuiltBackChannel(m);
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
@Override
public SOAPMessage call(SOAPMessage msgOut, Object addressObject) throws SOAPException 
{
   checkClosed();
   
   String address = getAddress(addressObject);
   ConduitInitiator ci = getConduitInitiator(address);
    
    
   // create a new Message and Exchange
   EndpointInfo info = new EndpointInfo();
   info.setAddress(address);
   Message outMessage = new MessageImpl();
   Exchange exch = new ExchangeImpl();
   outMessage.setExchange(exch);
   exch.put("org.apache.cxf.transport.process_fault_on_http_400", true); //JBWS-3945
    
   // sent SOAPMessage
   try 
   {
      final Conduit c = ci.getConduit(info, BusFactory.getThreadDefaultBus(false)); //TODO verify bus

      if (msgOut.saveRequired())
      {
         msgOut.saveChanges();
      }
    
      Map<String, List<String>> outHeaders = new HashMap<String, List<String>>();
      for (Iterator<?> it = msgOut.getMimeHeaders().getAllHeaders(); it.hasNext();) 
      {
         MimeHeader mimeHeader = (MimeHeader)it.next();
         if ("Content-Type".equals(mimeHeader.getName())) 
         {
            outMessage.put(Message.CONTENT_TYPE, mimeHeader.getValue());
         }
            
         // disable the chunked encoding if requested
         if ("Transfer-Encoding".equals(mimeHeader.getName())
             && "disabled".equals(mimeHeader.getValue())
             && c instanceof HTTPConduit) 
         {
            ((HTTPConduit)c).getClient().setAllowChunking(false);
            continue;
         }
            
         List<String> values = outHeaders.get(mimeHeader.getName());
         if (values == null) 
         {
            values = new ArrayList<String>();
            outHeaders.put(mimeHeader.getName(), values);
         } 
         values.add(mimeHeader.getValue());
      }
      outMessage.put(Message.HTTP_REQUEST_METHOD, "POST");
      outMessage.put(Message.PROTOCOL_HEADERS, outHeaders);
      c.prepare(outMessage);
        
      OutputStream outs = outMessage.getContent(OutputStream.class);
      msgOut.writeTo(outs);
        
      c.setMessageObserver(createMessageObserver(c));
        
      c.close(outMessage);
   } 
   catch (Exception ex) 
   {
      throw MESSAGES.soapMessageCouldNotBeSent(ex);
   }    

   // read SOAPMessage        
   return readSoapMessage(exch); 
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
@Override
public SOAPMessage get(Object addressObject) throws SOAPException 
{
   checkClosed();
   
   String address = getAddress(addressObject);
   ConduitInitiator ci = getConduitInitiator(address);
    
    
   // create a new Message and Exchange
   EndpointInfo info = new EndpointInfo();
   info.setAddress(address);
   Message outMessage = new MessageImpl();
   Exchange exch = new ExchangeImpl();
   outMessage.setExchange(exch);
    
   // sent GET request
   try 
   {
      final Conduit c = ci.getConduit(info, BusFactory.getThreadDefaultBus(false)); //TODO verify bus
      
      if (c instanceof HTTPConduit)
      {
         ((HTTPConduit)c).getClient().setAutoRedirect(true);
      }
        
      outMessage.put(Message.HTTP_REQUEST_METHOD, "GET");
      c.prepare(outMessage);
        
      c.setMessageObserver(createMessageObserver(c));
        
      c.close(outMessage);
   } 
   catch (Exception ex) 
   {
      throw MESSAGES.getRequestCouldNotBeSent(ex);
   }    

   // read SOAPMessage
   return readSoapMessage(exch);
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
@Override
protected Conduit getInbuiltBackChannel(Message inMessage) {
    return new BackChannelConduit(EndpointReferenceUtils.getAnonymousEndpointReference(), inMessage);
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
/**
 * Called on completion of the MEP for which the Conduit was required.
 * 
 * @param exchange
 *            represents the completed MEP
 */
public void complete(Exchange exchange) {
	InvocationKey key = new InvocationKey(exchange);
	InvocationContext invocation = getInvocation(key);
	boolean failover = false;
	Conduit old = (Conduit) exchange.getOutMessage().remove(Conduit.class.getName());
	if (requiresFailover(exchange)) {
		onFailure(invocation.getContext());
		LOG.debug("Failover {}", invocation.getContext());
		Endpoint failoverTarget = getFailoverTarget(exchange, invocation);
		if (failoverTarget != null) {
			setEndpoint(failoverTarget);
			if (old != null) {
				old.close();
				conduits.remove(old);
			}
			failover = performFailover(exchange, invocation);
		}
	} else {
		if (invocation != null) {
			onSuccess(invocation.getContext());
		}
	}
	if (!failover) {
		LOG.debug("Failover not required");
		synchronized (this) {
			inProgress.remove(key);
		}
		if (MessageUtils.isTrue(exchange.get("KeepConduitAlive"))) {
			return;
		}

		try {
			if (exchange.getInMessage() != null) {
				Conduit c = (Conduit) exchange.getOutMessage().get(Conduit.class);
				if (c == null) {
					getSelectedConduit(exchange.getInMessage()).close(exchange.getInMessage());
				} else {
					c.close(exchange.getInMessage());
				}
			}
		} catch (IOException e) {
		}
	}
}
 

import org.apache.cxf.transport.Conduit; //导入依赖的package包/类
private Conduit getConduit(Message inMessage) throws IOException {
    Exchange exchange = inMessage.getExchange();
    Conduit conduit = exchange.getDestination().getBackChannel(inMessage);
    exchange.setConduit(conduit);
    return conduit;
}