|
|
RequirementsWe should support to deploy cxf client(as service consumer) and cxf server (as service provider) easilly into servicemix. key componentscxf service engineoverviewServiceMix Cxf Se component is a JBI Service Engine exposing (annotated) POJO as services on the JBI Bus. features for cxf service engine
main class for cxf service engineCxfSeComponentWe use maven-xbean-plugin to generate component xbean schema, so we need add comment to specify the xml element name firstly
/**
*
* @org.apache.xbean.XBean element="component"
*/
This class just extends servicemix DefaultComponent
@Override
protected void doInit() throws Exception {
bus = BusFactory.getDefaultBus();
super.doInit();
}
CxfSeEndpointWe use maven-xbean-plugin to generate endpoint xbean schema, so we need add comment to specify the xml element name firstly
/**
*
* @org.apache.xbean.XBean element="endpoint"
*/
In this class we init and publish jaxws endpoint with cxf JBI transport and JBI binding, and also register this endpoint to servicemix.
/*
* (non-Javadoc)
*
* @see org.apache.servicemix.common.Endpoint#validate()
*/
@Override
public void validate() throws DeploymentException {
if (getPojo() == null) {
throw new DeploymentException("pojo must be set");
}
JaxWsServiceFactoryBean serviceFactory = new JaxWsServiceFactoryBean();
serviceFactory.setPopulateFromClass(true);
endpoint = new EndpointImpl(getBus(), getPojo(),
new JaxWsServerFactoryBean(serviceFactory));
endpoint.setBindingUri(org.apache.cxf.binding.jbi.JBIConstants.NS_JBI_BINDING);//using cxf jbi binding, so the message use JBI wrapper
endpoint.setInInterceptors(getInInterceptors());
endpoint.setInFaultInterceptors(getInFaultInterceptors());
endpoint.setOutInterceptors(getOutInterceptors());
endpoint.setOutFaultInterceptors(getOutFaultInterceptors());
endpoint.getOutInterceptors().add(new WrapperOutInterceptor());
if (isMtomEnabled()) {
endpoint.getInInterceptors().add(new AttachmentInInterceptor());
}
JaxWsImplementorInfo implInfo = new JaxWsImplementorInfo(getPojo()
.getClass());
setService(implInfo.getServiceName()); // register servicename to NMR
setInterfaceName(implInfo.getInterfaceName()); // register interfacename(portType) to NMR
setEndpoint(implInfo.getEndpointName().getLocalPart()); // register Endpoint(Port) name to NMR
super.validate();
}
We publish the endpoint when endpoint start in servicemix
/*
* (non-Javadoc)
*
* @see org.apache.servicemix.common.endpoints.ProviderEndpoint#start()
*/
@Override
public void start() throws Exception {
super.start();
address = "jbi://" + ID_GENERATOR.generateSanitizedId();//use jbi prefix for the endpoint address to pick up cxf JBI transport
endpoint.publish(address);
setService(endpoint.getServer().getEndpoint().getService().getName());
setEndpoint(endpoint.getServer().getEndpoint().getEndpointInfo()
.getName().getLocalPart());
}
We just get MessageExchange from Servicemix NMR and delegate it to cxf JBI transport
/*
* (non-Javadoc)
*
* @see org.apache.servicemix.common.endpoints.ProviderEndpoint#process(javax.jbi.messaging.MessageExchange)
*/
@Override
public void process(MessageExchange exchange) throws Exception {
JBITransportFactory jbiTransportFactory = (JBITransportFactory) getBus()
.getExtension(ConduitInitiatorManager.class)
.getConduitInitiator(CxfSeComponent.JBI_TRANSPORT_ID);//load JBITransportFactory from cxf bus
QName serviceName = exchange.getService();
if (serviceName == null) {
serviceName = getService();
exchange.setService(serviceName);
}
QName interfaceName = exchange.getInterfaceName();
if (interfaceName == null) {
interfaceName = getInterfaceName();
exchange.setInterfaceName(interfaceName);
}
JBIDestination jbiDestination = jbiTransportFactory
.getDestination(serviceName.toString()
+ interfaceName.toString());// each endpoint should have its own JBIDestination, so just query it by serviceName and interfaceName
DeliveryChannel dc = getContext().getDeliveryChannel();
jbiTransportFactory.setDeliveryChannel(dc);
jbiDestination.setDeliveryChannel(dc);
if (exchange.getStatus() == ExchangeStatus.ACTIVE) {
jbiDestination.getJBIDispatcherUtil().dispatch(exchange);//delegate MessageExchange to cxf JBI transport, cxf in charge of message unmarshalling and service invocation
}
}
CxfSeProxyFactoryBeanWe use maven-xbean-plugin to generate proxy xbean schema, so we need add comment to specify the xml element name firstly /** * * @org.apache.xbean.XBean element="proxy" description="A CXF proxy" * */ We create client proxy exactly the same as we do in cxf
private Object createProxy() throws Exception {
JaxWsProxyFactoryBean cf = new JaxWsProxyFactoryBean();
cf.setServiceName(getService());
cf.setServiceClass(type);
cf.setAddress("jbi://" + new IdGenerator().generateSanitizedId());//use jbi prefix for the endpoint address to pick up cxf JBI transport
cf.setBindingId(org.apache.cxf.binding.jbi.JBIConstants.NS_JBI_BINDING);//use cxf JBI binding
Bus bus = BusFactory.getDefaultBus();
JBITransportFactory jbiTransportFactory = (JBITransportFactory) bus
.getExtension(ConduitInitiatorManager.class)
.getConduitInitiator(CxfSeComponent.JBI_TRANSPORT_ID);
if (getInternalContext() != null) {
DeliveryChannel dc = getInternalContext().getDeliveryChannel();
if (dc != null) {
jbiTransportFactory.setDeliveryChannel(dc);
}
}
return cf.create();
}
configureationinterceptors configurationSince cxfse is using apache cxf internally, so customer can configure cxf se endpoint with inteceptors which follow cxf inteceptor api.
<cxfse:endpoint>
<cxfse:pojo>
<bean class="org.apache.cxf.calculator.CalculatorImpl">
</bean>
</cxfse:pojo>
<cxfse:inInterceptors>
<bean class="org.apache.cxf.interceptor.LoggingInInterceptor"/>
</cxfse:inInterceptors>
<cxfse:outInterceptors>
<bean class="org.apache.cxf.interceptor.LoggingOutInterceptor"/>
</cxfse:outInterceptors>
<cxfse:inFaultInterceptors>
<bean class="org.apache.cxf.interceptor.LoggingInInterceptor"/>
</cxfse:inFaultInterceptors>
<cxfse:outFaultInterceptors>
<bean class="org.apache.cxf.interceptor.LoggingOutInterceptor"/>
</cxfse:outFaultInterceptors>
</cxfse:endpoint>
proxy configurationCustomer can use proxy from one of the client bean, or from inside another component, and call the JBI endpoint as a plain Java object.
<bean class="org.apache.servicemix.cxfse.GreeterImplForClientProxy">
<property name="calculator">
<cxfse:proxy service="calculator:CalculatorService" type="org.apache.cxf.calculator.CalculatorPortType" />
</property>
</bean>
cxf binding componentoverviewServiceMix ships with a JBI compliant HTTP/SOAP or JMS/SOAP binding component named servicemix-cxf-bc which use apache cxf internally. features for cxf binding component
main class for cxf binding componentCxfBcComponentWe use maven-xbean-plugin to generate component xbean schema, so we need add comment to specify the xml element name firstly
/**
*
* @org.apache.xbean.XBean element="component"
*/
This class just extends servicemix DefaultComponent
@Override
protected void doInit() throws Exception {
if (getBusConfig() != null) {//for the cxf bc, we need pass in cxf bus configuration file sometimes, e.g. we need configure bus to support ws-*
SpringBusFactory bf = new SpringBusFactory();
bus = bf.createBus(getBusConfig());
} else {
bus = BusFactory.newInstance().createBus();
}
super.doInit();
}
CxfBcConsumerWe use maven-xbean-plugin to generate consumer xbean schema, so we need add comment to specify the xml element name firstly
/**
*
* @org.apache.xbean.XBean element="consumer"
*/
Consumer of binding component means expose JBI internal endpoint to outside client, for example
CxfBcProvider
/**
*
* @org.apache.xbean.XBean element="provider"
*/
Provider of binding component means bridge internal request to outside server, for example
The second half in CxfBcProviderMessageObserver
public void onMessage(Message message) {
try {
...
...
inChain.add(providerEndpoint.getInInterceptors());
inChain.add(providerEndpoint.getInFaultInterceptors());
soapMessage.setInterceptorChain(inChain);
inChain.doIntercept(soapMessage);//transform soap message to jbi message
//send back jbi message to nmr
if (boi.getOperationInfo().isOneWay()) {
messageExchange.setStatus(ExchangeStatus.DONE);
} else if (soapMessage.get("jbiFault") != null
&& soapMessage.get("jbiFault").equals(true)) {
...
...
} else {
messageExchange.setStatus(ExchangeStatus.DONE);
}
boolean txSync = messageExchange.getStatus() == ExchangeStatus.ACTIVE
&& messageExchange.isTransacted()
&& Boolean.TRUE.equals(messageExchange
.getProperty(JbiConstants.SEND_SYNC));
if (txSync) {
providerEndpoint.getContext().getDeliveryChannel().sendSync(
messageExchange);
} else {
providerEndpoint.getContext().getDeliveryChannel().send(
messageExchange);
}
} catch (Exception e) {
e.printStackTrace();
}
...
...
}
configurationSince cxfbc is using apache cxf internally, so customer can configure cxf bc provider or consumer with inteceptors which follow cxf inteceptor api. A full example typical use caseWe have test cases almost cover all features of cxf bc and se, such as
|