hi all
I m one of the members of SCI-flex (Flexible Integration of CEP and
SOA)team. We have currently implemented XML and AXIOM mediator in case of
connecting Esper with Synapse. Now i m going to write a test for the XML
mediator. it is kind of tricky as i am new to synapse.
In our mediator we are setting esper configuration , epl statement
settings etc. Here i am messed bit wheather i have to test each and every of
those etc.
Can anyone help me on how i should approach this. How i test configuration
settings etc.
I have attached the source of XML mediator below ,it would be help full i
guess
public class XMLMediator implements Mediator, ManagedLifecycle {
/**
* Log associated with the XML Mediator.
*/
private static final Log log = LogFactory.getLog(XMLMediator.class);
/**
* Trace State.
*/
private int traceState = 0;
/**
* The Instance URI to be used by EPServiceProviderManager.
* @see com.espertech.esper.client.EPServiceProviderManager
*/
private String instanceURI = null;
/**
* Esper Configuration instance.
*/
private Configuration configuration = new Configuration();
/**
* Synapse Listener Instance.
*/
private SynapseListener listener = null;
/**
* Helper to handle EPL statements.
*/
private EPLStatementHelper eplStatementHelper = null;
/**
* Sets the EventToAddress. Please set listener before setting the
* EventToAddress.
* @param uri URI of To Address associated with Event.
*/
public void setEventToAddress(String uri) {
if (listener != null)
listener.setEventToAddress(uri);
else
log.error("Listener has not been set");
}
/**
* Sets the Esper Configuration details from an Axiom Element describing
* the various details.
* @param config Configuration Axiomimport java.lang.reflect Element.
*/
public void setConfiguration(OMElement config) {
log.debug("Setting configuration " + config);
try {
Document doc = DocumentBuilderFactory.newInstance()
.newDocumentBuilder().parse(
new ByteArrayInputStream(config.toString().getBytes()));
configuration.configure(doc);
log.info("Setting configuration complete");
} catch (Exception e) {
log.error("An error occured while setting the configuration "
+ e.getMessage());
}
}
/**
* Gets the Trace State.
* @return Trace State.
*/
public int getTraceState() {
return traceState;
}
/**
* Sets the Trace State.
* @param state Trace State.
*/
public void setTraceState(int state) {
traceState = state;
}
/**
* Sets the Instance URI.
* @param uri the Instance URI.
*/
public void setInstanceURI(String uri) {
log.debug("Setting Instance URI " + uri);
instanceURI = uri;
}
/**
* Sets associated listener.
* @param name name of listener class.
*/
public void setListener(String name) {
log.debug("Setting listener " + name);
try {
Class listenerClass = Class.forName(name);
if (listenerClass == null) {
log.error("Invalid Class Name");
return;
}
Object o = listenerClass.newInstance();
if (o instanceof SynapseListener) {
listener = (SynapseListener) o;
log.info("Listener " + name + " was successfully setup");
}
else
log.error("Setting listener failed");
} catch (ClassNotFoundException e) {
log.error("Class " + name + " was not found");
} catch (Exception e) {
log.error("Setting listener failed " + e.getMessage());
}
}
/**
* Sets the EPL Statement.
* @param epl the EPL Statement element.
*/
public void setStatement(OMElement epl) {
if (listener == null) {
listener = new SynapseListenerImpl();
}
log.debug("Setting EPL statement " + epl);
String value = epl.getAttributeValue(new QName("value"));
if (value == null) {
value = epl.getAttributeValue(new QName("key"));
if (value == null) {
log.error("Setting EPL statement failed. Got " + epl);
return;
}
else {
log.debug("Setting EPL statment using registry key " +
value);
eplStatementHelper = new EPLStatementHelper(
EPLStatementHelper.EPLStatementType.INDIRECT, value,
getProvider(), listener);
}
}
else {
log.debug("Setting EPL statement " + value);
eplStatementHelper = new EPLStatementHelper(value,
getProvider(), listener);
}
}
/**
* Sets the EPL Statement.
* @param epl the EPL Statement.
*/
public void setStatement(String epl) {
if (listener == null) {
listener = new SynapseListenerImpl();
}
log.debug("Setting EPL statement " + epl);
eplStatementHelper = new EPLStatementHelper(epl, getProvider(),
listener);
}
/**
* Gets an EPServiceProvider based on Mediator configuration details.
* @return EPServiceProvider instance.
*/
private EPServiceProvider getProvider() {
try {
if (instanceURI == null && configuration == null)
return EPServiceProviderManager.getDefaultProvider();
else if (instanceURI == null)
return EPServiceProviderManager
.getDefaultProvider(configuration);
else if (configuration == null)
return EPServiceProviderManager.getProvider(instanceURI);
else
return EPServiceProviderManager.getProvider(instanceURI,
configuration);
} catch (ConfigurationException e) {
log.error("An error occured while retrieving provider "
+ e.getMessage());
return null;
}
}
/**
* Exposes the type of the current mediator for logging and debugging
* purposes.
* @return Mediator Type.
*/
public String getType() {
return "org.sciflex.plugins.synapse.esper.mediators.XMLMediator";
}
/**
* Invokes the mediator passing the current message for mediation.
* @param mc Message Context of the current message.
* @return returns true if mediation should continue, or false if
further
* mediation should be aborted.
*/
public boolean mediate(MessageContext mc) {
log.trace("Beginning Mediation");
EPServiceProvider provider = getProvider();
if (provider == null) {
/*
* There should be an error if we couldn't obtain the provider.
* Therefore, stop mediation
*/
return false;
}
eplStatementHelper.invoke(mc);
OMElement bodyElement = null;
try {
bodyElement = mc.getEnvelope().getBody().getFirstElement();
} catch (OMException e) {
log.warn("An error occured while reading message "
+ e.getMessage());
// We don't mind an error while reading a message.
return true;
}
if (bodyElement == null) {
// FIXME Figure out proper response for null element.
return true;
}
bodyElement.build();
try {
String buffer = bodyElement.toStringWithConsume();
XMLStreamReader xsr = XMLInputFactory.newInstance()
.createXMLStreamReader(new ByteArrayInputStream(
buffer.getBytes()));
StAXOMBuilder builder = new StAXOMBuilder(DOOMAbstractFactory
.getOMFactory(), xsr);
OMElement docElement = builder.getDocumentElement();
if (docElement != null) {
Node node = (Node) docElement.getParent();
provider.getEPRuntime().sendEvent(node);
log.trace("Ending Mediation");
}
else
log.error("Mediation failed");
} catch (Exception e) {
log.error("An error occured while sending Event " +
e.getMessage());
}
return true;
}
/**
* Destruction of Initialized resources.
*/
public void destroy() {
// We have nothing to destroy.
}
/**
* Initialization of resources.
* @param se Synapse Environment passed by configuration.
*/
public void init(SynapseEnvironment se) {
// hack to get round lack of init(SynapseEnv) on mediator interface.
if (listener != null)
listener.setSynapseEnvironment(se);
else
log.error("Listener has not been set");
}
}
Honestly i am really new to those. Hence your responses are highly admired.
Thanking you
kind regards
Harsha
