You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@synapse.apache.org by "Wright, Peter" <Pe...@six-group.com> on 2013/09/05 14:34:23 UTC
RE: Synapse: TCP syslog to JMS proxy: messages only processed in
synapse after client TCP connection is closed
Hi Hiranja,
I've rewritten my class mediator to just read the message content according to the message lengths,
and changed the log mediator settings as you suggested, but there still appears to be a problem.
When a client writes 3 separate messages, opening and closing the TCP connection each time, all messages
are received and processed correctly by synapse.
However, when a client opens a TCP connection, writes 3 separate messages, and then closes the connection,
only the first message is received and processed correctly by synapse.
My class mediator processes the first message correctly, so it appears(?) to be processing and returning
the message content correctly (without blocking), but there are no log entries at all for the following
messages.
I've attached my various config files and the source code of my class mediator.
Thanks for your help,
Peter
======================================
synapse.log
-----------
2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,MediatorLog.java:97,To: , WSAction: urn:mediate, SOAPAction: urn:mediate, MessageID: urn:uuid:B3133E5C6DF9C641551378383293591, Direction: request
2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:78,===syslogMsgBuilder.mediate called=============
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:161,getSyslogMsgs. start
2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:171,getSyslogMsgs. Reading msg 1 =============
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:229,readMsgLen. start
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.0=1
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.1=9
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.2=4
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:260,readMsgLen: strMsglen = '194'
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:283,readMsgLen: intMsglen = 194
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:299,readMsgContent. start. readLen=194
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:342,readMsgContent: msgContent=[<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] ], nlFound=false
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:195,getSyslogMsgs: no NL found: stop
2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:85,Found 1 message(s). Delete existing payload
2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:105,syslogMsg.1='<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] '
2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:113,xmlStr.1='<text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text>'
2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:132,Made 1 message(s)
2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:136,SOAP.Envelope: <?xml version='1.0' encoding='utf-8'?><soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"><soapenv:Body><text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text></soapenv:Body></soapenv:Envelope>
synapse.xml
-----------
<definitions xmlns="http://ws.apache.org/ns/synapse">
<sequence name="fault">
<makefault>
<code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
<reason value="Mediation failed."/>
</makefault>
<send/>
</sequence>
<sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
<in>
<log level="simple"/>
<send/>
</in>
<out>
<send/>
</out>
</sequence>
<proxy name="proxyTcp2Jms" transports="tcp">
<target>
<inSequence>
<!-- Define TCP listener expects plain text (syslog) messages -->
<property name="messageType" value="text/plain" scope="axis2"/>
<property name="OUT_ONLY" value="true"/>
<property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
<log level="simple"/> <!-- simple,headers,full -->
<!-- Split any multiple syslog messages for the iterator -->
<!-- and return as XML with child messages -->
<class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
<!-- <log level="full"/> -->
</class>
<!-- <log level="full"/> -->
<!-- Iterate over any multiple messages -->
<iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
<target>
<sequence>
<send>
<!-- <log level="full"/> -->
<endpoint>
<address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
</endpoint>
</send>
</sequence>
</target>
</iterate>
</inSequence>
<!-- OUT_ONLY=true
<outSequence>
<log level="simple" category="INFO" separator=","/>
<send/>
</outSequence>
-->
<faultSequence>
<log level="simple" category="ERROR" separator=","/>
</faultSequence>
</target>
<parameter name="transport.tcp.port">6060</parameter>
<parameter name="transport.tcp.contentType">text/plain</parameter>
<!-- <parameter name="transport.tcp.backlog">0</parameter> -->
</proxy>
</definitions>
syslogMsgBuilder.java
---------------------
/*
* Filename: syslogMsgBuilder.java
* Author: P.Wright
* Date: 06.08.2013
* Description: Java class for Synapse class mediator.
* Class is embedded in Synapse, and called when a SYSLOG message is received.
*
* The SYSLOG message can contain 1-n syslog messages, each
* separated by a NL, and the length of each message defined
* at the start of it. This is called TCP-Framing "octet-counted".
* See: http://www.rsyslog.com/doc/omfwd.html
*
* Example input syslog message containing 3 messages
* 35 <1> This is dummy message number 1\n
* 35 <2> This is dummy message number 2\n
* 34 <3> This is dummy message number 3
*
* Example Synapse-SOAP message with 3 syslog messages in the XML payload:
* <?xml version='1.0' encoding='utf-8'?>
* <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
* <soapenv:Body>
* <text xmlns="http://ws.apache.org/commons/ns/payload">
* 35 <1> This is dummy message number 1\n
* 35 <2> This is dummy message number 2\n
* 34 <3> This is dummy message number 3
* </text>
* </soapenv:Body>
* </soapenv:Envelope>
*
* Actions as follows:
* - Get the first element of the message body
* - Get a reader to the content of the first element
* - Read the msglen
* - Read the message according to msglen
* - If last character is NOT a '\n', stop reading messages
* - If last character IS a '\n', keep going and read next msglen
* - Replace any "<" with "<"
* - Replace the payload with an XML content with 3 messages (see below)
*
* Output message as follows:
* <?xml version='1.0' encoding='utf-8'?>
* <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
* <soapenv:Body>
* <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
* <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
* <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
* </soapenv:Body>
* </soapenv:Envelope>
*
* Method Overview:
* - mediate ("main" method)
*/
package com.sixtelekurs.classMediators;
import java.io.Reader;
import java.util.Iterator;
import java.util.ArrayList;
import java.util.List;
import org.apache.axiom.om.OMElement;
import org.apache.axiom.om.util.AXIOMUtil;
import org.apache.axiom.soap.SOAPBody;
import org.apache.axiom.soap.SOAPEnvelope;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.synapse.MessageContext;
import org.apache.synapse.Mediator;
public class syslogMsgBuilder implements Mediator
{
private static String className = "syslogMsgBuilder";
private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
private String _desc = null;
private String _uid = null;
// ------------------------------------------------------------------------
public boolean mediate(MessageContext mc)
{
String logMsg = "===" + className + ".mediate called=============";
log.info(logMsg);
// Get the message content as an array of strings
List<String> listSyslogStr = getSyslogMsgs(mc.getEnvelope().getBody().getFirstElement());
// Delete the existing payload
logMsg = "Found " + listSyslogStr.size() + " message(s). Delete existing payload";
log.info(logMsg);
SOAPBody soapBody = mc.getEnvelope().getBody();
for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
{
itr.next();
itr.remove();
}
// Traverse the string list, convert each to an XML element
// and write into the SOAP body
int msgCnt = 0;
try
{
// Get the syslog msg
for (String msg : listSyslogStr)
{
// Replace all "<" with "<" for XML
String syslogMsg = msg.replaceAll("<", "<");
int jj = msgCnt+1;
logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
log.debug(logMsg);
String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
xmlStr += syslogMsg;
xmlStr += "</text>";
msgCnt++;
OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
log.info(logMsg);
soapBody.addChild(xmlElem);
}
}
catch (Exception e)
{
logMsg = "ERROR: " + e;
log.error(logMsg);
return(false);
}
if (msgCnt == 0)
{
logMsg = "ERROR: no messages processed";
log.error(logMsg);
return(false);
}
logMsg = "Made " + msgCnt + " message(s)";
log.info(logMsg);
SOAPEnvelope envelope = mc.getEnvelope();
logMsg = "SOAP.Envelope: " + envelope;
log.info(logMsg);
return(true);
}
class ReturnValues
{
String msgString;
boolean foundNewline;
ReturnValues(final String msgString, boolean foundNewline)
{
this.msgString = msgString;
this.foundNewline = foundNewline;
}
}
// getSyslogMsgs: traverses the input message
// and returns any syslog messages as an array
// of strings in XML form
private List<String> getSyslogMsgs(OMElement omElem)
{
String funcname = "getSyslogMsgs";
String logMsg = "";
logMsg = funcname + ". start";
log.debug(logMsg);
List<String> syslogMsgs = new ArrayList<String>();
Reader reader = omElem.getTextAsStream(false); // pass cache=false
int msgCnt = 0;
boolean keepGoing = true;
while (keepGoing)
{
logMsg = funcname + ". Reading msg " + (msgCnt+1) + " =============";
log.info(logMsg);
// Get the msglen
int msgLen = readMsgLen(reader);
if (msgLen <= 0)
{
keepGoing = false;
}
else
{
// Get the message
ReturnValues retvals = readMsgContent(reader, msgLen);
if (retvals.msgString.isEmpty())
{
keepGoing = false;
}
else
{
syslogMsgs.add(retvals.msgString);
// Read any trailing NL
if (!retvals.foundNewline)
{
logMsg = funcname + ": no NL found: stop";
log.debug(logMsg);
keepGoing = false;
}
}
}
msgCnt++;
}
// Close the reader
/*********************
try
{
reader.close();
}
catch (Exception e)
{
logMsg = funcname + ": reader.close failed. " + e;
log.error(logMsg);
}
*********************/
return(syslogMsgs);
}
// Get the msglen of the syslog message
// Read characters till a space and convert to an integer
// Return the int msglen
private int readMsgLen(Reader reader)
{
String funcname = "readMsgLen";
int intMsglen = 0;
String logMsg = "";
String strMsglen = "";
logMsg = funcname + ". start";
log.debug(logMsg);
try
{
char[] cBuffer = new char[10];
int cCnt=0;
boolean keepGoing = true;
while (keepGoing)
{
// Read one character at a time into the char array
int rc = reader.read(cBuffer, cCnt, 1);
if (rc <= 0)
{
keepGoing = false;
}
else if (cBuffer[cCnt] != ' ')
{
logMsg = "cBuffer." + cCnt + "=" + cBuffer[cCnt];
log.debug(logMsg);
cCnt++;
}
else
{
keepGoing = false;
}
}
if (cCnt > 0)
{
// Convert the characters read into a string
strMsglen = new String(cBuffer, 0, cCnt);
logMsg = funcname + ": strMsglen = '" + strMsglen + "'";
log.debug(logMsg);
}
}
catch (Exception e)
{
logMsg = funcname + ": error reading msglen" + e;
log.error(logMsg);
}
if (strMsglen != "")
{
// Convert the (string)msglen to an (int)msglen
try
{
intMsglen = Integer.parseInt(strMsglen);
}
catch (Exception e)
{
logMsg = funcname + ": error converting string '" + strMsglen + "' to integer. " + e;
log.error(logMsg);
}
}
logMsg = funcname + ": intMsglen = " + intMsglen;
log.debug(logMsg);
return(intMsglen);
}
// readMsgContent --------------------------------------------
private ReturnValues readMsgContent(Reader reader, int msgLen)
{
String funcname = "readMsgContent";
String msgContent = "";
boolean nlFound = false;
String logMsg = "";
int maxlen = 2048;
char[] cBuffer = new char[maxlen];
int readLen = (msgLen < (maxlen-1) ? msgLen:(maxlen-1));
logMsg = funcname + ". start. readLen=" + readLen;
log.debug(logMsg);
if (msgLen > maxlen-1)
{
logMsg = "message too long for buffer (" + msgLen + "/" + (maxlen-1);
log.error(logMsg);
ReturnValues retvals = new ReturnValues("", false);
return(retvals);
}
try
{
int rc = reader.read(cBuffer, 0, readLen);
if (rc <= 0)
{
logMsg = funcname + ": error reading msg: " + rc;
log.error(logMsg);
}
else
{
if (rc < readLen)
{
logMsg = "read: IST=" + rc + ", SOLL=" + readLen;
log.warn(logMsg);
}
// Ignore any trailing NL
if (cBuffer[rc-1] == '\n')
{
nlFound = true;
logMsg = "NL read at end of message - ignore it";
log.debug(logMsg);
rc--;
}
msgContent = new String(cBuffer, 0, rc);
}
}
catch (Exception e)
{
logMsg = funcname + ": exception reading msg" + e;
log.error(logMsg);
}
logMsg = funcname + ": msgContent=[" + msgContent + "], nlFound=" + nlFound;
log.debug(logMsg);
ReturnValues retvals = new ReturnValues(msgContent, nlFound);
return(retvals);
}
// SET methods --------------------------------
public String getDescription()
{
String msg = className + ".desc";
return(msg);
}
public String getType()
{
String msg = className + ".type";
return(msg);
}
public int getTraceState()
{
return(0);
}
// SET methods --------------------------------
public void setTraceState(int traceState)
{
traceState = 0;
}
public void setDescription(String desc)
{
_desc = desc;
}
public void setUid(String uid)
{
_uid = uid;
}
}
-----Original Message-----
From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
Sent: Freitag, 30. August 2013 19:11
To: user@synapse.apache.org
Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
On Aug 30, 2013, at 5:40 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
> Hi Hiranja,
>
> I've done a series of tests that appear to indicate the problem
> is in the TCP transport.
It's not really a problem in the tcp transport. The issue is exactly what I mentioned in my previous mail. That is, Synapse tries to read the input stream all the way to the end-of-stream, which causes it to get blocked. In addition to your custom mediator, other mediators like <lov level="full"/> also cause the same behavior. Basically any mediator that accesses the full message payload can cause this problem.
The way I see it, you have only three options at this point:
1. Change your class mediator to not read the message payload to the end-of-stream. But you still have to get rid of any other mediators that may cause the message to be fully serialized (e.g. <log level="full"/>). Therefore this may not be very practical.
2. Write a custom message builder instead of a class mediator. This is probably the most architecturally sound approach. However, you must make sure that the builder doesn't read the input stream all the way to the end-of-stream. That is, it should read the message up to a known delimiter, or read a known number of bytes. Otherwise you will have the same problem again.
3. Use HTTP to receive the messages (HTTP protocol handles this type of issues using the Content-Length header or the chunked encoding system).
> Here are the steps I did and the results I saw:
> Important: "still the same" means:
> - while the tcp connection is open, there is no reaction from synapse (no log messages etc)
> - only AFTER the connection to the TCP listener has been closed do log messages appear and the syslog messages are processed
>
> 1. Rewrote my class mediator to use BufferedReader.readLine() -> still the same
> 2. Rewrote my class mediator to simply replace the payload with fixed messages WITHOUT even reading the existing payload -> still the same
> 3. Commented out my class mediator in synapse.xml -> still the same
All the above 3 are mainly due to the other serializing mediators in the configuration (e.g. log). Also FYI, changing to the BufferedRead.readLine() doesn't make any difference. That is also a blocking I/O call.
> 4. Added transport.tcp.port to the axis2.xml (as well as being in proxyTcp2Jms in synapse.xml) -> error at synapse startup "Error while starting the TCP endpoint. Address already in use"
> 5. As above, but removed the port from proxyTcp2Jms in synapse.xml -> error at synapse startup "Service doesn't have configuration information for transport tcp"
> 6. Defined transport.tcp.port in both files, but with different port numbers (6060 in synapse.xml, 6061 in axis2.xml) -> synapse starts up OK (with two logfile entries):
> - TCPServer.java:76,TCP server started on port : 6061
> - TCPServer.java:76,TCP server started on port : 6060
> -> when I send a syslog message to port 6060 -> still the same
> -> when I send a syslog message to port 6061 -> error written to logfile immediately: AxisEngine.java:219,The service cannot be found for the endpoint reference (EPR)
> It appears that the TCPserver configured in synapse.xml is blocking until the client closes the connection,
> and the TCPListener configured in axis2.xml is not blocking, but cannot find the proxy definition.
That's not how it works. Messages sent to port 6061 must be dispatched by looking at the first element of the SOAP body. In your case it's just a generic <text/> element. Therefore Synapse fails to dispatch the message to any proxy service. This happens before any of the mediators are invoked. Therefore Synapse doesn't get to access the full message payload in this case. Hence it fails early in the pipeline, before Synapse gets blocked on I/O.
Thanks,
Hiranya
>
> The tcp jar file I'm using is this one:
> http://maven.wso2.org/nexus/content/groups/wso2-public/org/apache/axis2/axis2-transport-tcp/1.1.0-SNAPSHOT/axis2-transport-tcp-1.1.0-SNAPSHOT.jar
> but today also tried this one but saw no difference:
> http://dist.wso2.org/maven2//org/apache/axis2/axis2-transport-tcp/1.1.0-wso2v5/axis2-transport-tcp-1.1.0-wso2v5.jar
>
> Thanks,
> Peter
>
> -----Original Message-----
> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
> Sent: Montag, 26. August 2013 19:44
> To: user@synapse.apache.org
> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>
> Hi,
>
> On Aug 26, 2013, at 4:48 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>
>> Hi Hiranja,
>>
>> No problem. Thanks for the support.
>>
>> Peter
>>
>> ===============================================================
>>
>> /*
>> * Filename: syslogMsgBuilder.java
>> * Author: P.Wright
>> * Date: 06.08.2013
>> * Description: Java class for Synapse class mediator.
>> * Class is embedded in Synapse, and called when a SYSLOG message is received.
>> *
>> * The SYSLOG message can contain 1-n syslog messages, each
>> * separated by a NL, and the length of each message defined
>> * at the start of it. This is called TCP-Framing "octet-counted".
>> * See: http://www.rsyslog.com/doc/omfwd.html
>> *
>> * Example input syslog message containing 3 messages
>> * 34 <1> This is dummy message number 1
>> * 34 <2> This is dummy message number 2
>> * 34 <3> This is dummy message number 3
>> *
>> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
>> * <?xml version='1.0' encoding='utf-8'?>
>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>> * <soapenv:Body>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload">
>> * 34 <1> This is dummy message number 1
>> * 34 <2> This is dummy message number 2
>> * 34 <3> This is dummy message number 3
>> * </text>
>> * </soapenv:Body>
>> * </soapenv:Envelope>
>> *
>> * Actions as follows:
>> * - Get the payload
>> * - Split the payload on NL
>> * - Remove the msglen at the start of each message
>> * - Replace any "<" with "<"
>> * - Replace the payload with an XML content with 3 messages (see below)
>> *
>> * Output message as follows:
>> * <?xml version='1.0' encoding='utf-8'?>
>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>> * <soapenv:Body>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
>> * </soapenv:Body>
>> * </soapenv:Envelope>
>> *
>> * Method Overview:
>> * - mediate ("main" method)
>> */
>> package com.sixtelekurs.classMediators;
>>
>> import java.util.Iterator;
>> import org.apache.axiom.om.OMElement;
>> import org.apache.axiom.om.util.AXIOMUtil;
>> import org.apache.axiom.soap.SOAPBody;
>> import org.apache.axiom.soap.SOAPEnvelope;
>> import org.apache.commons.logging.Log;
>> import org.apache.commons.logging.LogFactory;
>> import org.apache.synapse.MessageContext;
>> import org.apache.synapse.Mediator;
>>
>>
>> public class syslogMsgBuilder implements Mediator
>> {
>> private static String className = "syslogMsgBuilder";
>> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
>> private String _desc = null;
>> private String _uid = null;
>>
>> // ------------------------------------------------------------------------
>> public boolean mediate(MessageContext mc)
>> {
>> String logMsg = "===" + className + ".mediate called=============";
>> log.debug(logMsg);
>>
>> // Get the message content and process it
>> String syslogStr = mc.getEnvelope().getBody().getFirstElement().getText();
>
> I believe this line if the issue. The call to getText() on the OMSourcedElement probably causes it to read the entire input stream until end-of-stream is encountered. But end-of-stream does not occur until the sender closes its output stream at the remote end.
>
> I think a proper solution should get the InputStream from the OMSourcedElement and try to read from it until a known number of bytes have been read or a particular delimiter character is encountered. In a case like HTTP, the HTTP message usually have the content-length header or a special delimiter sequence (chunk delimiter) which allows the server to know how much data to read from the input stream without reading all the way to the end-of-stream. In case of TCP transport, there are no special headers to look at, so you must rely on the information available on the message payload it self.
>
> That's just my take on it. May be there's a better/easier way to solve this problem at Axiom level (perhaps a special method in the OMSourcedElement API?). May be Andreas can shed some light on the matter.
>
> Thanks,
> Hiranya
>
>> logMsg = "syslogStr='" + syslogStr + "'";
>> log.debug(logMsg);
>>
>> // Delete the existing payload
>> logMsg = "Deleting existing payload";
>> log.debug(logMsg);
>> SOAPBody soapBody = mc.getEnvelope().getBody();
>> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
>> {
>> itr.next();
>> itr.remove();
>> }
>>
>> // Split the strings into N syslog messages, convert each to
>> // an OMElement, and append to the XML-Body
>> int msgCnt = 0;
>> try
>> {
>> // Split the string on NL
>> String[] lines = syslogStr.split("\\n");
>> logMsg = "syslog.input.cnt(NL)=" + lines.length;
>> log.debug(logMsg);
>> for (int ii=0; ii < lines.length; ii++)
>> {
>> // Now split on space (once only) to ignore the msglen
>> String[] msg = lines[ii].split(" ", 2);
>> if (msg.length == 2)
>> {
>> // Replace all "<" with "<" for XML
>> String syslogMsg = msg[1].replaceAll("<", "<");
>> int jj = ii+1;
>> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
>> log.debug(logMsg);
>> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
>> xmlStr += syslogMsg;
>> xmlStr += "</text>";
>> msgCnt++;
>>
>> // Now add elem
>> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
>> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
>> log.debug(logMsg);
>> soapBody.addChild(xmlElem);
>> }
>> else
>> {
>> logMsg = "msg could not be split correctly [" + msg + "]";
>> log.error(logMsg);
>> }
>> }
>> }
>> catch (Exception e)
>> {
>> logMsg = "ERROR: " + e;
>> log.error(logMsg);
>> return(false);
>> }
>>
>> if (msgCnt == 0)
>> {
>> logMsg = "ERROR: no messages processed";
>> log.error(logMsg);
>> return(false);
>> }
>>
>> logMsg = "Made " + msgCnt + " messages";
>> log.info(logMsg);
>>
>> SOAPEnvelope envelope = mc.getEnvelope();
>> logMsg = "SOAP.Envelope: " + envelope;
>> log.debug(logMsg);
>>
>> return(true);
>> }
>>
>> // SET methods --------------------------------
>> public String getDescription()
>> {
>> String msg = className + ".desc";
>> return(msg);
>> }
>> public String getType()
>> {
>> String msg = className + ".type";
>> return(msg);
>> }
>> public int getTraceState()
>> {
>> return(0);
>> }
>>
>> // SET methods --------------------------------
>> public void setTraceState(int traceState)
>> {
>> traceState = 0;
>> }
>> public void setDescription(String desc)
>> {
>> _desc = desc;
>> }
>> public void setUid(String uid)
>> {
>> _uid = uid;
>> }
>> }
>>
>> -----Original Message-----
>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>> Sent: Freitag, 23. August 2013 08:30
>> To: user@synapse.apache.org
>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>
>> It sounds like something is trying to read from the input stream until the end of stream is encountered. This may be the plain text builder or your custom mediator. Can you share your custom mediator source code so we can take a look?
>>
>> Thanks,
>> Hiranya
>>
>> On Aug 22, 2013, at 8:27 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>
>>> Hi,
>>>
>>> I have my TCP Syslog --> JMS proxy up and running (see previous emails), but now have encountered another problem.
>>> Up till now I have simply tested by using netstat to send 1-n messages stored in a file to the TCP server.
>>> In this case, the messages are being received and processed correctly by synapse (sent as JMS
>>> messages to the JMS server). It seems that this is working correctly, because the connection to the
>>> synapse TCP server is being closed by netstat after it sends the messages.
>>>
>>> In my real world case however, we have 1-n clients writing syslog messages to a
>>> local syslog server. This syslog server has been configured to send the messages on
>>> to the (remote) synapse TCP server. In this case the messages are NOT being received
>>> and processed by synapse. Only after the syslog server closes it connection to the synapse
>>> TCP server, are the messages being received and processed by synapse.
>>>
>>> Any ideas what the problem could be?
>>> Attached again my synapse.xml config file.
>>>
>>> Thanks,
>>> Peter
>>>
>>> ===============================================
>>>
>>> <definitions xmlns="http://ws.apache.org/ns/synapse">
>>> <sequence name="fault">
>>> <makefault>
>>> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
>>> <reason value="Mediation failed."/>
>>> </makefault>
>>> <send/>
>>> </sequence>
>>> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
>>> <in>
>>> <log level="full"/>
>>> <send/>
>>> </in>
>>> <out>
>>> <send/>
>>> </out>
>>> </sequence>
>>>
>>> <proxy name="proxyTcp2Jms" transports="tcp">
>>> <target>
>>> <inSequence>
>>> <!-- Define TCP listener expects plain text (syslog) messages -->
>>> <property name="messageType" value="text/plain" scope="axis2"/>
>>> <property name="OUT_ONLY" value="true"/>
>>> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
>>> <log level="full"/>
>>>
>>> <!-- Split any multiple syslog messages for the iterator -->
>>> <!-- and return as XML with child messages -->
>>> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
>>> <log level="full"/>
>>> </class>
>>>
>>> <log level="full"/>
>>>
>>> <!-- Iterate over any multiple messages -->
>>> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
>>> <target>
>>> <sequence>
>>> <send>
>>> <endpoint>
>>> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
>>> </endpoint>
>>> </send>
>>> </sequence>
>>> </target>
>>> </iterate>
>>> </inSequence>
>>> <outSequence/>
>>> <faultSequence>
>>> <log level="full" category="ERROR" separator=","/>
>>> </faultSequence>
>>> </target>
>>> <parameter name="transport.tcp.port">6060</parameter>
>>> <parameter name="transport.tcp.contentType">text/plain</parameter>
>>> </proxy>
>>>
>>> </definitions>
>>>
>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>> Thank you.
>>
>> --
>> Hiranya Jayathilaka
>> Mayhem Lab/RACE Lab;
>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>> Blog: http://techfeast-hiranya.blogspot.com
>>
>> The content of this e-mail is intended only for the confidential use of the person addressed.
>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>> Thank you.
>
> --
> Hiranya Jayathilaka
> Mayhem Lab/RACE Lab;
> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
> Blog: http://techfeast-hiranya.blogspot.com
>
> The content of this e-mail is intended only for the confidential use of the person addressed.
> If you are not the intended recipient, please notify the sender and delete this email immediately.
> Thank you.
--
Hiranya Jayathilaka
Mayhem Lab/RACE Lab;
Dept. of Computer Science, UCSB; http://cs.ucsb.edu
E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
Blog: http://techfeast-hiranya.blogspot.com
The content of this e-mail is intended only for the confidential use of the person addressed.
If you are not the intended recipient, please notify the sender and delete this email immediately.
Thank you.
RE: Synapse: TCP syslog to JMS proxy: messages only processed in
synapse after client TCP connection is closed
Posted by "Wright, Peter" <Pe...@six-group.com>.
Hi Andreas,
Thanks for the input that confirms my test results about the persistent connections.
Pity that my "solution" is not recommended.
Initial tests OK, but of course no guarantee...
I've gone ahead and also implemented the UDP transport and will give both
to the test team and make a decision later about replacing the TCP
solution with the pipe transport.
Best regards,
Peter
-----Original Message-----
From: Andreas Veithen [mailto:andreas.veithen@gmail.com]
Sent: Montag, 16. September 2013 18:41
To: user@synapse.apache.org
Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
The TCP transport assumes that a connection is used for a single
message, i.e. messages are delimited by connection termination. If you
attempt to use that transport with a persistent connection to receive
multiple messages, then from the point of view of the transport,
Axis2, Axiom and Synapse, this will look like a single message with
infinite length. This then also means that the mediator(s) processing
this "message" will remain active indefinitely. This is something
neither the transport nor Synapse has been designed for. Since the
entire stack supports streaming, you might get this working, but you
will likely run into all kinds of problems because your use case falls
completely outside of what Synapse has been designed for.
I think that you only have three realistic options to make this work:
1. Use UDP.
2. Use the pipe transport [1]. This transport was actually designed
for the syslog use case (but for an older version of the syslog
protocol). This should be simple to set up if your syslog daemon and
Synapse run on the same Unix host. Just create a named pipe, configure
syslog to write to that pipe and configure Synapse to listen to the
pipe. However, you may need to write some code to support the expected
syslog protocol version. If Synapse runs on a distinct host, you still
have the option to set up a syslog daemon on that host and to
configure the other one to forward syslog messages. Finally, Synapse
runs on Windows, then this won't work because the pipe transport only
supports Unix named pipes.
3. Write your own TCP transport by starting from the pipe transport.
Andreas
[1] https://synapse.apache.org/apidocs/org/apache/synapse/transport/pipe/package-summary.html
On Wed, Sep 11, 2013 at 2:11 PM, Wright, Peter
<Pe...@six-group.com> wrote:
> Hi Hiranja,
>
> I've done some more tests and have seen that my class can only work if
> the client (in our case a syslog server configured to forward messages
> to the synapse tcp listener) actually regularly opens/closes its connection
> to the tcp listener.
>
> Only when the connection is closed does my class mediator get all sent messages.
> As long as the connection is open, my read blocks and my class mediator
> cannot complete and pass the messages onto the synapse iterator mediator.
>
> If I return before the connection is closed, all messages read up till
> then will be sent on correctly to the iterator mediator, and will then
> be forwarded as JMS messages as desired, but any remaining messages
> sent by the syslog client will be lost.
>
> The problem is that synapse works sequentially, but in this case an
> asynchronous (callback?) type of mediator would be necessary to allow
> the handling of each message as it is read.
>
> I'm thinking of using the tcp listener and my class mediator, but
> removing the iterator mediator. I could extend my class to directly
> send each message as a JMS message to the messaging server (in java code).
> My class would remain active as long as the tcp client keeps the connection open.
> That way I could at least use the synapse multi threaded tcp listener
> and not have to write that myself.
>
> What do you think? Any other suggestions?
>
> Thanks,
> Peter
>
> -----Original Message-----
> From: Wright, Peter
> Sent: Dienstag, 10. September 2013 16:25
> To: 'user@synapse.apache.org'
> Subject: RE: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>
> Hi Hiranja,
>
> I'm still having a problem with finding "end of synapse message".
>
> A synapse message can contain 1-n syslog messages.
> Each syslog message is preceeded by its message length,
> so I have no problem reading each message.
>
> The problem is to know when to stop trying to read a possible
> next message i.e. end of synapse message (end of frame?).
> When I try to read one byte of the length of the next
> (non existing) message, the read blocks.
>
> Every message has a terminating newline character, which
> is not necessary, as the message length is defined anyway.
> If the newline were only at the end of frame (synapse message),
> I could use that to know I should stop reading.
>
> Who is adding the newlines? The syslog sender, or the TCP transport
> module of synapse? Any chance of finding a solution here?
> (end of frame delimiter?).
>
>> TCP transport will inject all messages sent on the same connection as a single message into Synapse.
> Does the TCP transport buffer the input syslog messages before concatenating them into the synapse message?
> When are the messages sent? Buffer size reached? Time? Hopefully not only after the TCP connection is closed??
> Is one synapse message equivalent to one TCP frame?
>
> In our test setup, we have a syslog server configured to send all received messages
> on to the synapse TCP listener.
>
> Thanks
> Peter
>
> -----Original Message-----
> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
> Sent: Donnerstag, 5. September 2013 19:59
> To: user@synapse.apache.org
> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>
> You need to somehow figure out the number of messages to read from the input stream and read all those messages in your custom mediator. TCP transport will inject all messages sent on the same connection as a single message into Synapse.
>
> If you want to debug your code, start synapse with the flag -xdebug and do a remote debug via port 8000.
>
> Thanks,
> Hiranya
>
> On Sep 5, 2013, at 5:34 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>
>> Hi Hiranja,
>>
>> I've rewritten my class mediator to just read the message content according to the message lengths,
>> and changed the log mediator settings as you suggested, but there still appears to be a problem.
>>
>> When a client writes 3 separate messages, opening and closing the TCP connection each time, all messages
>> are received and processed correctly by synapse.
>> However, when a client opens a TCP connection, writes 3 separate messages, and then closes the connection,
>> only the first message is received and processed correctly by synapse.
>>
>> My class mediator processes the first message correctly, so it appears(?) to be processing and returning
>> the message content correctly (without blocking), but there are no log entries at all for the following
>> messages.
>>
>> I've attached my various config files and the source code of my class mediator.
>>
>> Thanks for your help,
>> Peter
>>
>> ======================================
>>
>>
>> synapse.log
>> -----------
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,MediatorLog.java:97,To: , WSAction: urn:mediate, SOAPAction: urn:mediate, MessageID: urn:uuid:B3133E5C6DF9C641551378383293591, Direction: request
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:78,===syslogMsgBuilder.mediate called=============
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:161,getSyslogMsgs. start
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:171,getSyslogMsgs. Reading msg 1 =============
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:229,readMsgLen. start
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.0=1
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.1=9
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.2=4
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:260,readMsgLen: strMsglen = '194'
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:283,readMsgLen: intMsglen = 194
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:299,readMsgContent. start. readLen=194
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:342,readMsgContent: msgContent=[<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] ], nlFound=false
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:195,getSyslogMsgs: no NL found: stop
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:85,Found 1 message(s). Delete existing payload
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:105,syslogMsg.1='<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] '
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:113,xmlStr.1='<text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text>'
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:132,Made 1 message(s)
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:136,SOAP.Envelope: <?xml version='1.0' encoding='utf-8'?><soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"><soapenv:Body><text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text></soapenv:Body></soapenv:Envelope>
>>
>> synapse.xml
>> -----------
>> <definitions xmlns="http://ws.apache.org/ns/synapse">
>> <sequence name="fault">
>> <makefault>
>> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
>> <reason value="Mediation failed."/>
>> </makefault>
>> <send/>
>> </sequence>
>> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
>> <in>
>> <log level="simple"/>
>> <send/>
>> </in>
>> <out>
>> <send/>
>> </out>
>> </sequence>
>>
>> <proxy name="proxyTcp2Jms" transports="tcp">
>> <target>
>> <inSequence>
>> <!-- Define TCP listener expects plain text (syslog) messages -->
>> <property name="messageType" value="text/plain" scope="axis2"/>
>> <property name="OUT_ONLY" value="true"/>
>> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
>> <log level="simple"/> <!-- simple,headers,full -->
>>
>> <!-- Split any multiple syslog messages for the iterator -->
>> <!-- and return as XML with child messages -->
>> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
>> <!-- <log level="full"/> -->
>> </class>
>>
>> <!-- <log level="full"/> -->
>>
>> <!-- Iterate over any multiple messages -->
>> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
>> <target>
>> <sequence>
>> <send>
>> <!-- <log level="full"/> -->
>> <endpoint>
>> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
>> </endpoint>
>> </send>
>> </sequence>
>> </target>
>> </iterate>
>> </inSequence>
>> <!-- OUT_ONLY=true
>> <outSequence>
>> <log level="simple" category="INFO" separator=","/>
>> <send/>
>> </outSequence>
>> -->
>> <faultSequence>
>> <log level="simple" category="ERROR" separator=","/>
>> </faultSequence>
>> </target>
>> <parameter name="transport.tcp.port">6060</parameter>
>> <parameter name="transport.tcp.contentType">text/plain</parameter>
>> <!-- <parameter name="transport.tcp.backlog">0</parameter> -->
>> </proxy>
>> </definitions>
>>
>>
>> syslogMsgBuilder.java
>> ---------------------
>> /*
>> * Filename: syslogMsgBuilder.java
>> * Author: P.Wright
>> * Date: 06.08.2013
>> * Description: Java class for Synapse class mediator.
>> * Class is embedded in Synapse, and called when a SYSLOG message is received.
>> *
>> * The SYSLOG message can contain 1-n syslog messages, each
>> * separated by a NL, and the length of each message defined
>> * at the start of it. This is called TCP-Framing "octet-counted".
>> * See: http://www.rsyslog.com/doc/omfwd.html
>> *
>> * Example input syslog message containing 3 messages
>> * 35 <1> This is dummy message number 1\n
>> * 35 <2> This is dummy message number 2\n
>> * 34 <3> This is dummy message number 3
>> *
>> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
>> * <?xml version='1.0' encoding='utf-8'?>
>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>> * <soapenv:Body>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload">
>> * 35 <1> This is dummy message number 1\n
>> * 35 <2> This is dummy message number 2\n
>> * 34 <3> This is dummy message number 3
>> * </text>
>> * </soapenv:Body>
>> * </soapenv:Envelope>
>> *
>> * Actions as follows:
>> * - Get the first element of the message body
>> * - Get a reader to the content of the first element
>> * - Read the msglen
>> * - Read the message according to msglen
>> * - If last character is NOT a '\n', stop reading messages
>> * - If last character IS a '\n', keep going and read next msglen
>> * - Replace any "<" with "<"
>> * - Replace the payload with an XML content with 3 messages (see below)
>> *
>> * Output message as follows:
>> * <?xml version='1.0' encoding='utf-8'?>
>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>> * <soapenv:Body>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
>> * </soapenv:Body>
>> * </soapenv:Envelope>
>> *
>> * Method Overview:
>> * - mediate ("main" method)
>> */
>>
>> package com.sixtelekurs.classMediators;
>>
>> import java.io.Reader;
>> import java.util.Iterator;
>> import java.util.ArrayList;
>> import java.util.List;
>> import org.apache.axiom.om.OMElement;
>> import org.apache.axiom.om.util.AXIOMUtil;
>> import org.apache.axiom.soap.SOAPBody;
>> import org.apache.axiom.soap.SOAPEnvelope;
>> import org.apache.commons.logging.Log;
>> import org.apache.commons.logging.LogFactory;
>> import org.apache.synapse.MessageContext;
>> import org.apache.synapse.Mediator;
>>
>>
>> public class syslogMsgBuilder implements Mediator
>> {
>> private static String className = "syslogMsgBuilder";
>> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
>> private String _desc = null;
>> private String _uid = null;
>>
>> // ------------------------------------------------------------------------
>> public boolean mediate(MessageContext mc)
>> {
>> String logMsg = "===" + className + ".mediate called=============";
>> log.info(logMsg);
>>
>> // Get the message content as an array of strings
>> List<String> listSyslogStr = getSyslogMsgs(mc.getEnvelope().getBody().getFirstElement());
>>
>> // Delete the existing payload
>> logMsg = "Found " + listSyslogStr.size() + " message(s). Delete existing payload";
>> log.info(logMsg);
>> SOAPBody soapBody = mc.getEnvelope().getBody();
>> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
>> {
>> itr.next();
>> itr.remove();
>> }
>>
>> // Traverse the string list, convert each to an XML element
>> // and write into the SOAP body
>> int msgCnt = 0;
>> try
>> {
>> // Get the syslog msg
>> for (String msg : listSyslogStr)
>> {
>> // Replace all "<" with "<" for XML
>> String syslogMsg = msg.replaceAll("<", "<");
>> int jj = msgCnt+1;
>> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
>> log.debug(logMsg);
>> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
>> xmlStr += syslogMsg;
>> xmlStr += "</text>";
>> msgCnt++;
>>
>> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
>> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
>> log.info(logMsg);
>> soapBody.addChild(xmlElem);
>> }
>> }
>> catch (Exception e)
>> {
>> logMsg = "ERROR: " + e;
>> log.error(logMsg);
>> return(false);
>> }
>>
>> if (msgCnt == 0)
>> {
>> logMsg = "ERROR: no messages processed";
>> log.error(logMsg);
>> return(false);
>> }
>>
>> logMsg = "Made " + msgCnt + " message(s)";
>> log.info(logMsg);
>>
>> SOAPEnvelope envelope = mc.getEnvelope();
>> logMsg = "SOAP.Envelope: " + envelope;
>> log.info(logMsg);
>>
>> return(true);
>> }
>>
>> class ReturnValues
>> {
>> String msgString;
>> boolean foundNewline;
>> ReturnValues(final String msgString, boolean foundNewline)
>> {
>> this.msgString = msgString;
>> this.foundNewline = foundNewline;
>> }
>> }
>>
>> // getSyslogMsgs: traverses the input message
>> // and returns any syslog messages as an array
>> // of strings in XML form
>> private List<String> getSyslogMsgs(OMElement omElem)
>> {
>> String funcname = "getSyslogMsgs";
>> String logMsg = "";
>>
>> logMsg = funcname + ". start";
>> log.debug(logMsg);
>>
>> List<String> syslogMsgs = new ArrayList<String>();
>> Reader reader = omElem.getTextAsStream(false); // pass cache=false
>>
>> int msgCnt = 0;
>> boolean keepGoing = true;
>> while (keepGoing)
>> {
>> logMsg = funcname + ". Reading msg " + (msgCnt+1) + " =============";
>> log.info(logMsg);
>>
>> // Get the msglen
>> int msgLen = readMsgLen(reader);
>> if (msgLen <= 0)
>> {
>> keepGoing = false;
>> }
>> else
>> {
>> // Get the message
>> ReturnValues retvals = readMsgContent(reader, msgLen);
>> if (retvals.msgString.isEmpty())
>> {
>> keepGoing = false;
>> }
>> else
>> {
>> syslogMsgs.add(retvals.msgString);
>>
>> // Read any trailing NL
>> if (!retvals.foundNewline)
>> {
>> logMsg = funcname + ": no NL found: stop";
>> log.debug(logMsg);
>> keepGoing = false;
>> }
>> }
>> }
>> msgCnt++;
>> }
>>
>> // Close the reader
>> /*********************
>> try
>> {
>> reader.close();
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": reader.close failed. " + e;
>> log.error(logMsg);
>> }
>> *********************/
>> return(syslogMsgs);
>> }
>>
>> // Get the msglen of the syslog message
>> // Read characters till a space and convert to an integer
>> // Return the int msglen
>> private int readMsgLen(Reader reader)
>> {
>> String funcname = "readMsgLen";
>> int intMsglen = 0;
>> String logMsg = "";
>> String strMsglen = "";
>>
>> logMsg = funcname + ". start";
>> log.debug(logMsg);
>>
>> try
>> {
>> char[] cBuffer = new char[10];
>> int cCnt=0;
>> boolean keepGoing = true;
>> while (keepGoing)
>> {
>> // Read one character at a time into the char array
>> int rc = reader.read(cBuffer, cCnt, 1);
>> if (rc <= 0)
>> {
>> keepGoing = false;
>> }
>> else if (cBuffer[cCnt] != ' ')
>> {
>> logMsg = "cBuffer." + cCnt + "=" + cBuffer[cCnt];
>> log.debug(logMsg);
>> cCnt++;
>> }
>> else
>> {
>> keepGoing = false;
>> }
>> }
>> if (cCnt > 0)
>> {
>> // Convert the characters read into a string
>> strMsglen = new String(cBuffer, 0, cCnt);
>> logMsg = funcname + ": strMsglen = '" + strMsglen + "'";
>> log.debug(logMsg);
>> }
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": error reading msglen" + e;
>> log.error(logMsg);
>> }
>>
>> if (strMsglen != "")
>> {
>> // Convert the (string)msglen to an (int)msglen
>> try
>> {
>> intMsglen = Integer.parseInt(strMsglen);
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": error converting string '" + strMsglen + "' to integer. " + e;
>> log.error(logMsg);
>> }
>> }
>> logMsg = funcname + ": intMsglen = " + intMsglen;
>> log.debug(logMsg);
>> return(intMsglen);
>> }
>>
>> // readMsgContent --------------------------------------------
>> private ReturnValues readMsgContent(Reader reader, int msgLen)
>> {
>> String funcname = "readMsgContent";
>> String msgContent = "";
>> boolean nlFound = false;
>> String logMsg = "";
>> int maxlen = 2048;
>> char[] cBuffer = new char[maxlen];
>> int readLen = (msgLen < (maxlen-1) ? msgLen:(maxlen-1));
>>
>> logMsg = funcname + ". start. readLen=" + readLen;
>> log.debug(logMsg);
>>
>> if (msgLen > maxlen-1)
>> {
>> logMsg = "message too long for buffer (" + msgLen + "/" + (maxlen-1);
>> log.error(logMsg);
>> ReturnValues retvals = new ReturnValues("", false);
>> return(retvals);
>> }
>>
>> try
>> {
>> int rc = reader.read(cBuffer, 0, readLen);
>> if (rc <= 0)
>> {
>> logMsg = funcname + ": error reading msg: " + rc;
>> log.error(logMsg);
>> }
>> else
>> {
>> if (rc < readLen)
>> {
>> logMsg = "read: IST=" + rc + ", SOLL=" + readLen;
>> log.warn(logMsg);
>> }
>>
>> // Ignore any trailing NL
>> if (cBuffer[rc-1] == '\n')
>> {
>> nlFound = true;
>> logMsg = "NL read at end of message - ignore it";
>> log.debug(logMsg);
>> rc--;
>> }
>> msgContent = new String(cBuffer, 0, rc);
>> }
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": exception reading msg" + e;
>> log.error(logMsg);
>> }
>> logMsg = funcname + ": msgContent=[" + msgContent + "], nlFound=" + nlFound;
>> log.debug(logMsg);
>>
>> ReturnValues retvals = new ReturnValues(msgContent, nlFound);
>> return(retvals);
>> }
>>
>> // SET methods --------------------------------
>> public String getDescription()
>> {
>> String msg = className + ".desc";
>> return(msg);
>> }
>> public String getType()
>> {
>> String msg = className + ".type";
>> return(msg);
>> }
>> public int getTraceState()
>> {
>> return(0);
>> }
>>
>> // SET methods --------------------------------
>> public void setTraceState(int traceState)
>> {
>> traceState = 0;
>> }
>> public void setDescription(String desc)
>> {
>> _desc = desc;
>> }
>> public void setUid(String uid)
>> {
>> _uid = uid;
>> }
>> }
>>
>> -----Original Message-----
>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>> Sent: Freitag, 30. August 2013 19:11
>> To: user@synapse.apache.org
>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>
>>
>> On Aug 30, 2013, at 5:40 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>
>>> Hi Hiranja,
>>>
>>> I've done a series of tests that appear to indicate the problem
>>> is in the TCP transport.
>>
>> It's not really a problem in the tcp transport. The issue is exactly what I mentioned in my previous mail. That is, Synapse tries to read the input stream all the way to the end-of-stream, which causes it to get blocked. In addition to your custom mediator, other mediators like <lov level="full"/> also cause the same behavior. Basically any mediator that accesses the full message payload can cause this problem.
>>
>> The way I see it, you have only three options at this point:
>>
>> 1. Change your class mediator to not read the message payload to the end-of-stream. But you still have to get rid of any other mediators that may cause the message to be fully serialized (e.g. <log level="full"/>). Therefore this may not be very practical.
>> 2. Write a custom message builder instead of a class mediator. This is probably the most architecturally sound approach. However, you must make sure that the builder doesn't read the input stream all the way to the end-of-stream. That is, it should read the message up to a known delimiter, or read a known number of bytes. Otherwise you will have the same problem again.
>> 3. Use HTTP to receive the messages (HTTP protocol handles this type of issues using the Content-Length header or the chunked encoding system).
>>
>>> Here are the steps I did and the results I saw:
>>> Important: "still the same" means:
>>> - while the tcp connection is open, there is no reaction from synapse (no log messages etc)
>>> - only AFTER the connection to the TCP listener has been closed do log messages appear and the syslog messages are processed
>>>
>>> 1. Rewrote my class mediator to use BufferedReader.readLine() -> still the same
>>> 2. Rewrote my class mediator to simply replace the payload with fixed messages WITHOUT even reading the existing payload -> still the same
>>> 3. Commented out my class mediator in synapse.xml -> still the same
>>
>> All the above 3 are mainly due to the other serializing mediators in the configuration (e.g. log). Also FYI, changing to the BufferedRead.readLine() doesn't make any difference. That is also a blocking I/O call.
>>
>>> 4. Added transport.tcp.port to the axis2.xml (as well as being in proxyTcp2Jms in synapse.xml) -> error at synapse startup "Error while starting the TCP endpoint. Address already in use"
>>> 5. As above, but removed the port from proxyTcp2Jms in synapse.xml -> error at synapse startup "Service doesn't have configuration information for transport tcp"
>>> 6. Defined transport.tcp.port in both files, but with different port numbers (6060 in synapse.xml, 6061 in axis2.xml) -> synapse starts up OK (with two logfile entries):
>>> - TCPServer.java:76,TCP server started on port : 6061
>>> - TCPServer.java:76,TCP server started on port : 6060
>>> -> when I send a syslog message to port 6060 -> still the same
>>> -> when I send a syslog message to port 6061 -> error written to logfile immediately: AxisEngine.java:219,The service cannot be found for the endpoint reference (EPR)
>>> It appears that the TCPserver configured in synapse.xml is blocking until the client closes the connection,
>>> and the TCPListener configured in axis2.xml is not blocking, but cannot find the proxy definition.
>>
>> That's not how it works. Messages sent to port 6061 must be dispatched by looking at the first element of the SOAP body. In your case it's just a generic <text/> element. Therefore Synapse fails to dispatch the message to any proxy service. This happens before any of the mediators are invoked. Therefore Synapse doesn't get to access the full message payload in this case. Hence it fails early in the pipeline, before Synapse gets blocked on I/O.
>>
>> Thanks,
>> Hiranya
>>
>>>
>>> The tcp jar file I'm using is this one:
>>> http://maven.wso2.org/nexus/content/groups/wso2-public/org/apache/axis2/axis2-transport-tcp/1.1.0-SNAPSHOT/axis2-transport-tcp-1.1.0-SNAPSHOT.jar
>>> but today also tried this one but saw no difference:
>>> http://dist.wso2.org/maven2//org/apache/axis2/axis2-transport-tcp/1.1.0-wso2v5/axis2-transport-tcp-1.1.0-wso2v5.jar
>>>
>>> Thanks,
>>> Peter
>>>
>>> -----Original Message-----
>>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>>> Sent: Montag, 26. August 2013 19:44
>>> To: user@synapse.apache.org
>>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>>
>>> Hi,
>>>
>>> On Aug 26, 2013, at 4:48 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>>
>>>> Hi Hiranja,
>>>>
>>>> No problem. Thanks for the support.
>>>>
>>>> Peter
>>>>
>>>> ===============================================================
>>>>
>>>> /*
>>>> * Filename: syslogMsgBuilder.java
>>>> * Author: P.Wright
>>>> * Date: 06.08.2013
>>>> * Description: Java class for Synapse class mediator.
>>>> * Class is embedded in Synapse, and called when a SYSLOG message is received.
>>>> *
>>>> * The SYSLOG message can contain 1-n syslog messages, each
>>>> * separated by a NL, and the length of each message defined
>>>> * at the start of it. This is called TCP-Framing "octet-counted".
>>>> * See: http://www.rsyslog.com/doc/omfwd.html
>>>> *
>>>> * Example input syslog message containing 3 messages
>>>> * 34 <1> This is dummy message number 1
>>>> * 34 <2> This is dummy message number 2
>>>> * 34 <3> This is dummy message number 3
>>>> *
>>>> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
>>>> * <?xml version='1.0' encoding='utf-8'?>
>>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>>> * <soapenv:Body>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload">
>>>> * 34 <1> This is dummy message number 1
>>>> * 34 <2> This is dummy message number 2
>>>> * 34 <3> This is dummy message number 3
>>>> * </text>
>>>> * </soapenv:Body>
>>>> * </soapenv:Envelope>
>>>> *
>>>> * Actions as follows:
>>>> * - Get the payload
>>>> * - Split the payload on NL
>>>> * - Remove the msglen at the start of each message
>>>> * - Replace any "<" with "<"
>>>> * - Replace the payload with an XML content with 3 messages (see below)
>>>> *
>>>> * Output message as follows:
>>>> * <?xml version='1.0' encoding='utf-8'?>
>>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>>> * <soapenv:Body>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
>>>> * </soapenv:Body>
>>>> * </soapenv:Envelope>
>>>> *
>>>> * Method Overview:
>>>> * - mediate ("main" method)
>>>> */
>>>> package com.sixtelekurs.classMediators;
>>>>
>>>> import java.util.Iterator;
>>>> import org.apache.axiom.om.OMElement;
>>>> import org.apache.axiom.om.util.AXIOMUtil;
>>>> import org.apache.axiom.soap.SOAPBody;
>>>> import org.apache.axiom.soap.SOAPEnvelope;
>>>> import org.apache.commons.logging.Log;
>>>> import org.apache.commons.logging.LogFactory;
>>>> import org.apache.synapse.MessageContext;
>>>> import org.apache.synapse.Mediator;
>>>>
>>>>
>>>> public class syslogMsgBuilder implements Mediator
>>>> {
>>>> private static String className = "syslogMsgBuilder";
>>>> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
>>>> private String _desc = null;
>>>> private String _uid = null;
>>>>
>>>> // ------------------------------------------------------------------------
>>>> public boolean mediate(MessageContext mc)
>>>> {
>>>> String logMsg = "===" + className + ".mediate called=============";
>>>> log.debug(logMsg);
>>>>
>>>> // Get the message content and process it
>>>> String syslogStr = mc.getEnvelope().getBody().getFirstElement().getText();
>>>
>>> I believe this line if the issue. The call to getText() on the OMSourcedElement probably causes it to read the entire input stream until end-of-stream is encountered. But end-of-stream does not occur until the sender closes its output stream at the remote end.
>>>
>>> I think a proper solution should get the InputStream from the OMSourcedElement and try to read from it until a known number of bytes have been read or a particular delimiter character is encountered. In a case like HTTP, the HTTP message usually have the content-length header or a special delimiter sequence (chunk delimiter) which allows the server to know how much data to read from the input stream without reading all the way to the end-of-stream. In case of TCP transport, there are no special headers to look at, so you must rely on the information available on the message payload it self.
>>>
>>> That's just my take on it. May be there's a better/easier way to solve this problem at Axiom level (perhaps a special method in the OMSourcedElement API?). May be Andreas can shed some light on the matter.
>>>
>>> Thanks,
>>> Hiranya
>>>
>>>> logMsg = "syslogStr='" + syslogStr + "'";
>>>> log.debug(logMsg);
>>>>
>>>> // Delete the existing payload
>>>> logMsg = "Deleting existing payload";
>>>> log.debug(logMsg);
>>>> SOAPBody soapBody = mc.getEnvelope().getBody();
>>>> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
>>>> {
>>>> itr.next();
>>>> itr.remove();
>>>> }
>>>>
>>>> // Split the strings into N syslog messages, convert each to
>>>> // an OMElement, and append to the XML-Body
>>>> int msgCnt = 0;
>>>> try
>>>> {
>>>> // Split the string on NL
>>>> String[] lines = syslogStr.split("\\n");
>>>> logMsg = "syslog.input.cnt(NL)=" + lines.length;
>>>> log.debug(logMsg);
>>>> for (int ii=0; ii < lines.length; ii++)
>>>> {
>>>> // Now split on space (once only) to ignore the msglen
>>>> String[] msg = lines[ii].split(" ", 2);
>>>> if (msg.length == 2)
>>>> {
>>>> // Replace all "<" with "<" for XML
>>>> String syslogMsg = msg[1].replaceAll("<", "<");
>>>> int jj = ii+1;
>>>> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
>>>> log.debug(logMsg);
>>>> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
>>>> xmlStr += syslogMsg;
>>>> xmlStr += "</text>";
>>>> msgCnt++;
>>>>
>>>> // Now add elem
>>>> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
>>>> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
>>>> log.debug(logMsg);
>>>> soapBody.addChild(xmlElem);
>>>> }
>>>> else
>>>> {
>>>> logMsg = "msg could not be split correctly [" + msg + "]";
>>>> log.error(logMsg);
>>>> }
>>>> }
>>>> }
>>>> catch (Exception e)
>>>> {
>>>> logMsg = "ERROR: " + e;
>>>> log.error(logMsg);
>>>> return(false);
>>>> }
>>>>
>>>> if (msgCnt == 0)
>>>> {
>>>> logMsg = "ERROR: no messages processed";
>>>> log.error(logMsg);
>>>> return(false);
>>>> }
>>>>
>>>> logMsg = "Made " + msgCnt + " messages";
>>>> log.info(logMsg);
>>>>
>>>> SOAPEnvelope envelope = mc.getEnvelope();
>>>> logMsg = "SOAP.Envelope: " + envelope;
>>>> log.debug(logMsg);
>>>>
>>>> return(true);
>>>> }
>>>>
>>>> // SET methods --------------------------------
>>>> public String getDescription()
>>>> {
>>>> String msg = className + ".desc";
>>>> return(msg);
>>>> }
>>>> public String getType()
>>>> {
>>>> String msg = className + ".type";
>>>> return(msg);
>>>> }
>>>> public int getTraceState()
>>>> {
>>>> return(0);
>>>> }
>>>>
>>>> // SET methods --------------------------------
>>>> public void setTraceState(int traceState)
>>>> {
>>>> traceState = 0;
>>>> }
>>>> public void setDescription(String desc)
>>>> {
>>>> _desc = desc;
>>>> }
>>>> public void setUid(String uid)
>>>> {
>>>> _uid = uid;
>>>> }
>>>> }
>>>>
>>>> -----Original Message-----
>>>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>>>> Sent: Freitag, 23. August 2013 08:30
>>>> To: user@synapse.apache.org
>>>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>>>
>>>> It sounds like something is trying to read from the input stream until the end of stream is encountered. This may be the plain text builder or your custom mediator. Can you share your custom mediator source code so we can take a look?
>>>>
>>>> Thanks,
>>>> Hiranya
>>>>
>>>> On Aug 22, 2013, at 8:27 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>>>
>>>>> Hi,
>>>>>
>>>>> I have my TCP Syslog --> JMS proxy up and running (see previous emails), but now have encountered another problem.
>>>>> Up till now I have simply tested by using netstat to send 1-n messages stored in a file to the TCP server.
>>>>> In this case, the messages are being received and processed correctly by synapse (sent as JMS
>>>>> messages to the JMS server). It seems that this is working correctly, because the connection to the
>>>>> synapse TCP server is being closed by netstat after it sends the messages.
>>>>>
>>>>> In my real world case however, we have 1-n clients writing syslog messages to a
>>>>> local syslog server. This syslog server has been configured to send the messages on
>>>>> to the (remote) synapse TCP server. In this case the messages are NOT being received
>>>>> and processed by synapse. Only after the syslog server closes it connection to the synapse
>>>>> TCP server, are the messages being received and processed by synapse.
>>>>>
>>>>> Any ideas what the problem could be?
>>>>> Attached again my synapse.xml config file.
>>>>>
>>>>> Thanks,
>>>>> Peter
>>>>>
>>>>> ===============================================
>>>>>
>>>>> <definitions xmlns="http://ws.apache.org/ns/synapse">
>>>>> <sequence name="fault">
>>>>> <makefault>
>>>>> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
>>>>> <reason value="Mediation failed."/>
>>>>> </makefault>
>>>>> <send/>
>>>>> </sequence>
>>>>> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
>>>>> <in>
>>>>> <log level="full"/>
>>>>> <send/>
>>>>> </in>
>>>>> <out>
>>>>> <send/>
>>>>> </out>
>>>>> </sequence>
>>>>>
>>>>> <proxy name="proxyTcp2Jms" transports="tcp">
>>>>> <target>
>>>>> <inSequence>
>>>>> <!-- Define TCP listener expects plain text (syslog) messages -->
>>>>> <property name="messageType" value="text/plain" scope="axis2"/>
>>>>> <property name="OUT_ONLY" value="true"/>
>>>>> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
>>>>> <log level="full"/>
>>>>>
>>>>> <!-- Split any multiple syslog messages for the iterator -->
>>>>> <!-- and return as XML with child messages -->
>>>>> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
>>>>> <log level="full"/>
>>>>> </class>
>>>>>
>>>>> <log level="full"/>
>>>>>
>>>>> <!-- Iterate over any multiple messages -->
>>>>> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
>>>>> <target>
>>>>> <sequence>
>>>>> <send>
>>>>> <endpoint>
>>>>> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
>>>>> </endpoint>
>>>>> </send>
>>>>> </sequence>
>>>>> </target>
>>>>> </iterate>
>>>>> </inSequence>
>>>>> <outSequence/>
>>>>> <faultSequence>
>>>>> <log level="full" category="ERROR" separator=","/>
>>>>> </faultSequence>
>>>>> </target>
>>>>> <parameter name="transport.tcp.port">6060</parameter>
>>>>> <parameter name="transport.tcp.contentType">text/plain</parameter>
>>>>> </proxy>
>>>>>
>>>>> </definitions>
>>>>>
>>>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>>>> Thank you.
>>>>
>>>> --
>>>> Hiranya Jayathilaka
>>>> Mayhem Lab/RACE Lab;
>>>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>>>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>>>> Blog: http://techfeast-hiranya.blogspot.com
>>>>
>>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>>> Thank you.
>>>
>>> --
>>> Hiranya Jayathilaka
>>> Mayhem Lab/RACE Lab;
>>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>>> Blog: http://techfeast-hiranya.blogspot.com
>>>
>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>> Thank you.
>>
>> --
>> Hiranya Jayathilaka
>> Mayhem Lab/RACE Lab;
>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>> Blog: http://techfeast-hiranya.blogspot.com
>>
>> The content of this e-mail is intended only for the confidential use of the person addressed.
>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>> Thank you.
>
> --
> Hiranya Jayathilaka
> Mayhem Lab/RACE Lab;
> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
> Blog: http://techfeast-hiranya.blogspot.com
>
> The content of this e-mail is intended only for the confidential use of the person addressed.
> If you are not the intended recipient, please notify the sender and delete this email immediately.
> Thank you.
The content of this e-mail is intended only for the confidential use of the person addressed.
If you are not the intended recipient, please notify the sender and delete this email immediately.
Thank you.
Re: Synapse: TCP syslog to JMS proxy: messages only processed in
synapse after client TCP connection is closed
Posted by Andreas Veithen <an...@gmail.com>.
The TCP transport assumes that a connection is used for a single
message, i.e. messages are delimited by connection termination. If you
attempt to use that transport with a persistent connection to receive
multiple messages, then from the point of view of the transport,
Axis2, Axiom and Synapse, this will look like a single message with
infinite length. This then also means that the mediator(s) processing
this "message" will remain active indefinitely. This is something
neither the transport nor Synapse has been designed for. Since the
entire stack supports streaming, you might get this working, but you
will likely run into all kinds of problems because your use case falls
completely outside of what Synapse has been designed for.
I think that you only have three realistic options to make this work:
1. Use UDP.
2. Use the pipe transport [1]. This transport was actually designed
for the syslog use case (but for an older version of the syslog
protocol). This should be simple to set up if your syslog daemon and
Synapse run on the same Unix host. Just create a named pipe, configure
syslog to write to that pipe and configure Synapse to listen to the
pipe. However, you may need to write some code to support the expected
syslog protocol version. If Synapse runs on a distinct host, you still
have the option to set up a syslog daemon on that host and to
configure the other one to forward syslog messages. Finally, Synapse
runs on Windows, then this won't work because the pipe transport only
supports Unix named pipes.
3. Write your own TCP transport by starting from the pipe transport.
Andreas
[1] https://synapse.apache.org/apidocs/org/apache/synapse/transport/pipe/package-summary.html
On Wed, Sep 11, 2013 at 2:11 PM, Wright, Peter
<Pe...@six-group.com> wrote:
> Hi Hiranja,
>
> I've done some more tests and have seen that my class can only work if
> the client (in our case a syslog server configured to forward messages
> to the synapse tcp listener) actually regularly opens/closes its connection
> to the tcp listener.
>
> Only when the connection is closed does my class mediator get all sent messages.
> As long as the connection is open, my read blocks and my class mediator
> cannot complete and pass the messages onto the synapse iterator mediator.
>
> If I return before the connection is closed, all messages read up till
> then will be sent on correctly to the iterator mediator, and will then
> be forwarded as JMS messages as desired, but any remaining messages
> sent by the syslog client will be lost.
>
> The problem is that synapse works sequentially, but in this case an
> asynchronous (callback?) type of mediator would be necessary to allow
> the handling of each message as it is read.
>
> I'm thinking of using the tcp listener and my class mediator, but
> removing the iterator mediator. I could extend my class to directly
> send each message as a JMS message to the messaging server (in java code).
> My class would remain active as long as the tcp client keeps the connection open.
> That way I could at least use the synapse multi threaded tcp listener
> and not have to write that myself.
>
> What do you think? Any other suggestions?
>
> Thanks,
> Peter
>
> -----Original Message-----
> From: Wright, Peter
> Sent: Dienstag, 10. September 2013 16:25
> To: 'user@synapse.apache.org'
> Subject: RE: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>
> Hi Hiranja,
>
> I'm still having a problem with finding "end of synapse message".
>
> A synapse message can contain 1-n syslog messages.
> Each syslog message is preceeded by its message length,
> so I have no problem reading each message.
>
> The problem is to know when to stop trying to read a possible
> next message i.e. end of synapse message (end of frame?).
> When I try to read one byte of the length of the next
> (non existing) message, the read blocks.
>
> Every message has a terminating newline character, which
> is not necessary, as the message length is defined anyway.
> If the newline were only at the end of frame (synapse message),
> I could use that to know I should stop reading.
>
> Who is adding the newlines? The syslog sender, or the TCP transport
> module of synapse? Any chance of finding a solution here?
> (end of frame delimiter?).
>
>> TCP transport will inject all messages sent on the same connection as a single message into Synapse.
> Does the TCP transport buffer the input syslog messages before concatenating them into the synapse message?
> When are the messages sent? Buffer size reached? Time? Hopefully not only after the TCP connection is closed??
> Is one synapse message equivalent to one TCP frame?
>
> In our test setup, we have a syslog server configured to send all received messages
> on to the synapse TCP listener.
>
> Thanks
> Peter
>
> -----Original Message-----
> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
> Sent: Donnerstag, 5. September 2013 19:59
> To: user@synapse.apache.org
> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>
> You need to somehow figure out the number of messages to read from the input stream and read all those messages in your custom mediator. TCP transport will inject all messages sent on the same connection as a single message into Synapse.
>
> If you want to debug your code, start synapse with the flag -xdebug and do a remote debug via port 8000.
>
> Thanks,
> Hiranya
>
> On Sep 5, 2013, at 5:34 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>
>> Hi Hiranja,
>>
>> I've rewritten my class mediator to just read the message content according to the message lengths,
>> and changed the log mediator settings as you suggested, but there still appears to be a problem.
>>
>> When a client writes 3 separate messages, opening and closing the TCP connection each time, all messages
>> are received and processed correctly by synapse.
>> However, when a client opens a TCP connection, writes 3 separate messages, and then closes the connection,
>> only the first message is received and processed correctly by synapse.
>>
>> My class mediator processes the first message correctly, so it appears(?) to be processing and returning
>> the message content correctly (without blocking), but there are no log entries at all for the following
>> messages.
>>
>> I've attached my various config files and the source code of my class mediator.
>>
>> Thanks for your help,
>> Peter
>>
>> ======================================
>>
>>
>> synapse.log
>> -----------
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,MediatorLog.java:97,To: , WSAction: urn:mediate, SOAPAction: urn:mediate, MessageID: urn:uuid:B3133E5C6DF9C641551378383293591, Direction: request
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:78,===syslogMsgBuilder.mediate called=============
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:161,getSyslogMsgs. start
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:171,getSyslogMsgs. Reading msg 1 =============
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:229,readMsgLen. start
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.0=1
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.1=9
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.2=4
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:260,readMsgLen: strMsglen = '194'
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:283,readMsgLen: intMsglen = 194
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:299,readMsgContent. start. readLen=194
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:342,readMsgContent: msgContent=[<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] ], nlFound=false
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:195,getSyslogMsgs: no NL found: stop
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:85,Found 1 message(s). Delete existing payload
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:105,syslogMsg.1='<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] '
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:113,xmlStr.1='<text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text>'
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:132,Made 1 message(s)
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:136,SOAP.Envelope: <?xml version='1.0' encoding='utf-8'?><soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"><soapenv:Body><text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text></soapenv:Body></soapenv:Envelope>
>>
>> synapse.xml
>> -----------
>> <definitions xmlns="http://ws.apache.org/ns/synapse">
>> <sequence name="fault">
>> <makefault>
>> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
>> <reason value="Mediation failed."/>
>> </makefault>
>> <send/>
>> </sequence>
>> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
>> <in>
>> <log level="simple"/>
>> <send/>
>> </in>
>> <out>
>> <send/>
>> </out>
>> </sequence>
>>
>> <proxy name="proxyTcp2Jms" transports="tcp">
>> <target>
>> <inSequence>
>> <!-- Define TCP listener expects plain text (syslog) messages -->
>> <property name="messageType" value="text/plain" scope="axis2"/>
>> <property name="OUT_ONLY" value="true"/>
>> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
>> <log level="simple"/> <!-- simple,headers,full -->
>>
>> <!-- Split any multiple syslog messages for the iterator -->
>> <!-- and return as XML with child messages -->
>> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
>> <!-- <log level="full"/> -->
>> </class>
>>
>> <!-- <log level="full"/> -->
>>
>> <!-- Iterate over any multiple messages -->
>> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
>> <target>
>> <sequence>
>> <send>
>> <!-- <log level="full"/> -->
>> <endpoint>
>> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
>> </endpoint>
>> </send>
>> </sequence>
>> </target>
>> </iterate>
>> </inSequence>
>> <!-- OUT_ONLY=true
>> <outSequence>
>> <log level="simple" category="INFO" separator=","/>
>> <send/>
>> </outSequence>
>> -->
>> <faultSequence>
>> <log level="simple" category="ERROR" separator=","/>
>> </faultSequence>
>> </target>
>> <parameter name="transport.tcp.port">6060</parameter>
>> <parameter name="transport.tcp.contentType">text/plain</parameter>
>> <!-- <parameter name="transport.tcp.backlog">0</parameter> -->
>> </proxy>
>> </definitions>
>>
>>
>> syslogMsgBuilder.java
>> ---------------------
>> /*
>> * Filename: syslogMsgBuilder.java
>> * Author: P.Wright
>> * Date: 06.08.2013
>> * Description: Java class for Synapse class mediator.
>> * Class is embedded in Synapse, and called when a SYSLOG message is received.
>> *
>> * The SYSLOG message can contain 1-n syslog messages, each
>> * separated by a NL, and the length of each message defined
>> * at the start of it. This is called TCP-Framing "octet-counted".
>> * See: http://www.rsyslog.com/doc/omfwd.html
>> *
>> * Example input syslog message containing 3 messages
>> * 35 <1> This is dummy message number 1\n
>> * 35 <2> This is dummy message number 2\n
>> * 34 <3> This is dummy message number 3
>> *
>> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
>> * <?xml version='1.0' encoding='utf-8'?>
>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>> * <soapenv:Body>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload">
>> * 35 <1> This is dummy message number 1\n
>> * 35 <2> This is dummy message number 2\n
>> * 34 <3> This is dummy message number 3
>> * </text>
>> * </soapenv:Body>
>> * </soapenv:Envelope>
>> *
>> * Actions as follows:
>> * - Get the first element of the message body
>> * - Get a reader to the content of the first element
>> * - Read the msglen
>> * - Read the message according to msglen
>> * - If last character is NOT a '\n', stop reading messages
>> * - If last character IS a '\n', keep going and read next msglen
>> * - Replace any "<" with "<"
>> * - Replace the payload with an XML content with 3 messages (see below)
>> *
>> * Output message as follows:
>> * <?xml version='1.0' encoding='utf-8'?>
>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>> * <soapenv:Body>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
>> * </soapenv:Body>
>> * </soapenv:Envelope>
>> *
>> * Method Overview:
>> * - mediate ("main" method)
>> */
>>
>> package com.sixtelekurs.classMediators;
>>
>> import java.io.Reader;
>> import java.util.Iterator;
>> import java.util.ArrayList;
>> import java.util.List;
>> import org.apache.axiom.om.OMElement;
>> import org.apache.axiom.om.util.AXIOMUtil;
>> import org.apache.axiom.soap.SOAPBody;
>> import org.apache.axiom.soap.SOAPEnvelope;
>> import org.apache.commons.logging.Log;
>> import org.apache.commons.logging.LogFactory;
>> import org.apache.synapse.MessageContext;
>> import org.apache.synapse.Mediator;
>>
>>
>> public class syslogMsgBuilder implements Mediator
>> {
>> private static String className = "syslogMsgBuilder";
>> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
>> private String _desc = null;
>> private String _uid = null;
>>
>> // ------------------------------------------------------------------------
>> public boolean mediate(MessageContext mc)
>> {
>> String logMsg = "===" + className + ".mediate called=============";
>> log.info(logMsg);
>>
>> // Get the message content as an array of strings
>> List<String> listSyslogStr = getSyslogMsgs(mc.getEnvelope().getBody().getFirstElement());
>>
>> // Delete the existing payload
>> logMsg = "Found " + listSyslogStr.size() + " message(s). Delete existing payload";
>> log.info(logMsg);
>> SOAPBody soapBody = mc.getEnvelope().getBody();
>> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
>> {
>> itr.next();
>> itr.remove();
>> }
>>
>> // Traverse the string list, convert each to an XML element
>> // and write into the SOAP body
>> int msgCnt = 0;
>> try
>> {
>> // Get the syslog msg
>> for (String msg : listSyslogStr)
>> {
>> // Replace all "<" with "<" for XML
>> String syslogMsg = msg.replaceAll("<", "<");
>> int jj = msgCnt+1;
>> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
>> log.debug(logMsg);
>> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
>> xmlStr += syslogMsg;
>> xmlStr += "</text>";
>> msgCnt++;
>>
>> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
>> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
>> log.info(logMsg);
>> soapBody.addChild(xmlElem);
>> }
>> }
>> catch (Exception e)
>> {
>> logMsg = "ERROR: " + e;
>> log.error(logMsg);
>> return(false);
>> }
>>
>> if (msgCnt == 0)
>> {
>> logMsg = "ERROR: no messages processed";
>> log.error(logMsg);
>> return(false);
>> }
>>
>> logMsg = "Made " + msgCnt + " message(s)";
>> log.info(logMsg);
>>
>> SOAPEnvelope envelope = mc.getEnvelope();
>> logMsg = "SOAP.Envelope: " + envelope;
>> log.info(logMsg);
>>
>> return(true);
>> }
>>
>> class ReturnValues
>> {
>> String msgString;
>> boolean foundNewline;
>> ReturnValues(final String msgString, boolean foundNewline)
>> {
>> this.msgString = msgString;
>> this.foundNewline = foundNewline;
>> }
>> }
>>
>> // getSyslogMsgs: traverses the input message
>> // and returns any syslog messages as an array
>> // of strings in XML form
>> private List<String> getSyslogMsgs(OMElement omElem)
>> {
>> String funcname = "getSyslogMsgs";
>> String logMsg = "";
>>
>> logMsg = funcname + ". start";
>> log.debug(logMsg);
>>
>> List<String> syslogMsgs = new ArrayList<String>();
>> Reader reader = omElem.getTextAsStream(false); // pass cache=false
>>
>> int msgCnt = 0;
>> boolean keepGoing = true;
>> while (keepGoing)
>> {
>> logMsg = funcname + ". Reading msg " + (msgCnt+1) + " =============";
>> log.info(logMsg);
>>
>> // Get the msglen
>> int msgLen = readMsgLen(reader);
>> if (msgLen <= 0)
>> {
>> keepGoing = false;
>> }
>> else
>> {
>> // Get the message
>> ReturnValues retvals = readMsgContent(reader, msgLen);
>> if (retvals.msgString.isEmpty())
>> {
>> keepGoing = false;
>> }
>> else
>> {
>> syslogMsgs.add(retvals.msgString);
>>
>> // Read any trailing NL
>> if (!retvals.foundNewline)
>> {
>> logMsg = funcname + ": no NL found: stop";
>> log.debug(logMsg);
>> keepGoing = false;
>> }
>> }
>> }
>> msgCnt++;
>> }
>>
>> // Close the reader
>> /*********************
>> try
>> {
>> reader.close();
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": reader.close failed. " + e;
>> log.error(logMsg);
>> }
>> *********************/
>> return(syslogMsgs);
>> }
>>
>> // Get the msglen of the syslog message
>> // Read characters till a space and convert to an integer
>> // Return the int msglen
>> private int readMsgLen(Reader reader)
>> {
>> String funcname = "readMsgLen";
>> int intMsglen = 0;
>> String logMsg = "";
>> String strMsglen = "";
>>
>> logMsg = funcname + ". start";
>> log.debug(logMsg);
>>
>> try
>> {
>> char[] cBuffer = new char[10];
>> int cCnt=0;
>> boolean keepGoing = true;
>> while (keepGoing)
>> {
>> // Read one character at a time into the char array
>> int rc = reader.read(cBuffer, cCnt, 1);
>> if (rc <= 0)
>> {
>> keepGoing = false;
>> }
>> else if (cBuffer[cCnt] != ' ')
>> {
>> logMsg = "cBuffer." + cCnt + "=" + cBuffer[cCnt];
>> log.debug(logMsg);
>> cCnt++;
>> }
>> else
>> {
>> keepGoing = false;
>> }
>> }
>> if (cCnt > 0)
>> {
>> // Convert the characters read into a string
>> strMsglen = new String(cBuffer, 0, cCnt);
>> logMsg = funcname + ": strMsglen = '" + strMsglen + "'";
>> log.debug(logMsg);
>> }
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": error reading msglen" + e;
>> log.error(logMsg);
>> }
>>
>> if (strMsglen != "")
>> {
>> // Convert the (string)msglen to an (int)msglen
>> try
>> {
>> intMsglen = Integer.parseInt(strMsglen);
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": error converting string '" + strMsglen + "' to integer. " + e;
>> log.error(logMsg);
>> }
>> }
>> logMsg = funcname + ": intMsglen = " + intMsglen;
>> log.debug(logMsg);
>> return(intMsglen);
>> }
>>
>> // readMsgContent --------------------------------------------
>> private ReturnValues readMsgContent(Reader reader, int msgLen)
>> {
>> String funcname = "readMsgContent";
>> String msgContent = "";
>> boolean nlFound = false;
>> String logMsg = "";
>> int maxlen = 2048;
>> char[] cBuffer = new char[maxlen];
>> int readLen = (msgLen < (maxlen-1) ? msgLen:(maxlen-1));
>>
>> logMsg = funcname + ". start. readLen=" + readLen;
>> log.debug(logMsg);
>>
>> if (msgLen > maxlen-1)
>> {
>> logMsg = "message too long for buffer (" + msgLen + "/" + (maxlen-1);
>> log.error(logMsg);
>> ReturnValues retvals = new ReturnValues("", false);
>> return(retvals);
>> }
>>
>> try
>> {
>> int rc = reader.read(cBuffer, 0, readLen);
>> if (rc <= 0)
>> {
>> logMsg = funcname + ": error reading msg: " + rc;
>> log.error(logMsg);
>> }
>> else
>> {
>> if (rc < readLen)
>> {
>> logMsg = "read: IST=" + rc + ", SOLL=" + readLen;
>> log.warn(logMsg);
>> }
>>
>> // Ignore any trailing NL
>> if (cBuffer[rc-1] == '\n')
>> {
>> nlFound = true;
>> logMsg = "NL read at end of message - ignore it";
>> log.debug(logMsg);
>> rc--;
>> }
>> msgContent = new String(cBuffer, 0, rc);
>> }
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": exception reading msg" + e;
>> log.error(logMsg);
>> }
>> logMsg = funcname + ": msgContent=[" + msgContent + "], nlFound=" + nlFound;
>> log.debug(logMsg);
>>
>> ReturnValues retvals = new ReturnValues(msgContent, nlFound);
>> return(retvals);
>> }
>>
>> // SET methods --------------------------------
>> public String getDescription()
>> {
>> String msg = className + ".desc";
>> return(msg);
>> }
>> public String getType()
>> {
>> String msg = className + ".type";
>> return(msg);
>> }
>> public int getTraceState()
>> {
>> return(0);
>> }
>>
>> // SET methods --------------------------------
>> public void setTraceState(int traceState)
>> {
>> traceState = 0;
>> }
>> public void setDescription(String desc)
>> {
>> _desc = desc;
>> }
>> public void setUid(String uid)
>> {
>> _uid = uid;
>> }
>> }
>>
>> -----Original Message-----
>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>> Sent: Freitag, 30. August 2013 19:11
>> To: user@synapse.apache.org
>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>
>>
>> On Aug 30, 2013, at 5:40 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>
>>> Hi Hiranja,
>>>
>>> I've done a series of tests that appear to indicate the problem
>>> is in the TCP transport.
>>
>> It's not really a problem in the tcp transport. The issue is exactly what I mentioned in my previous mail. That is, Synapse tries to read the input stream all the way to the end-of-stream, which causes it to get blocked. In addition to your custom mediator, other mediators like <lov level="full"/> also cause the same behavior. Basically any mediator that accesses the full message payload can cause this problem.
>>
>> The way I see it, you have only three options at this point:
>>
>> 1. Change your class mediator to not read the message payload to the end-of-stream. But you still have to get rid of any other mediators that may cause the message to be fully serialized (e.g. <log level="full"/>). Therefore this may not be very practical.
>> 2. Write a custom message builder instead of a class mediator. This is probably the most architecturally sound approach. However, you must make sure that the builder doesn't read the input stream all the way to the end-of-stream. That is, it should read the message up to a known delimiter, or read a known number of bytes. Otherwise you will have the same problem again.
>> 3. Use HTTP to receive the messages (HTTP protocol handles this type of issues using the Content-Length header or the chunked encoding system).
>>
>>> Here are the steps I did and the results I saw:
>>> Important: "still the same" means:
>>> - while the tcp connection is open, there is no reaction from synapse (no log messages etc)
>>> - only AFTER the connection to the TCP listener has been closed do log messages appear and the syslog messages are processed
>>>
>>> 1. Rewrote my class mediator to use BufferedReader.readLine() -> still the same
>>> 2. Rewrote my class mediator to simply replace the payload with fixed messages WITHOUT even reading the existing payload -> still the same
>>> 3. Commented out my class mediator in synapse.xml -> still the same
>>
>> All the above 3 are mainly due to the other serializing mediators in the configuration (e.g. log). Also FYI, changing to the BufferedRead.readLine() doesn't make any difference. That is also a blocking I/O call.
>>
>>> 4. Added transport.tcp.port to the axis2.xml (as well as being in proxyTcp2Jms in synapse.xml) -> error at synapse startup "Error while starting the TCP endpoint. Address already in use"
>>> 5. As above, but removed the port from proxyTcp2Jms in synapse.xml -> error at synapse startup "Service doesn't have configuration information for transport tcp"
>>> 6. Defined transport.tcp.port in both files, but with different port numbers (6060 in synapse.xml, 6061 in axis2.xml) -> synapse starts up OK (with two logfile entries):
>>> - TCPServer.java:76,TCP server started on port : 6061
>>> - TCPServer.java:76,TCP server started on port : 6060
>>> -> when I send a syslog message to port 6060 -> still the same
>>> -> when I send a syslog message to port 6061 -> error written to logfile immediately: AxisEngine.java:219,The service cannot be found for the endpoint reference (EPR)
>>> It appears that the TCPserver configured in synapse.xml is blocking until the client closes the connection,
>>> and the TCPListener configured in axis2.xml is not blocking, but cannot find the proxy definition.
>>
>> That's not how it works. Messages sent to port 6061 must be dispatched by looking at the first element of the SOAP body. In your case it's just a generic <text/> element. Therefore Synapse fails to dispatch the message to any proxy service. This happens before any of the mediators are invoked. Therefore Synapse doesn't get to access the full message payload in this case. Hence it fails early in the pipeline, before Synapse gets blocked on I/O.
>>
>> Thanks,
>> Hiranya
>>
>>>
>>> The tcp jar file I'm using is this one:
>>> http://maven.wso2.org/nexus/content/groups/wso2-public/org/apache/axis2/axis2-transport-tcp/1.1.0-SNAPSHOT/axis2-transport-tcp-1.1.0-SNAPSHOT.jar
>>> but today also tried this one but saw no difference:
>>> http://dist.wso2.org/maven2//org/apache/axis2/axis2-transport-tcp/1.1.0-wso2v5/axis2-transport-tcp-1.1.0-wso2v5.jar
>>>
>>> Thanks,
>>> Peter
>>>
>>> -----Original Message-----
>>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>>> Sent: Montag, 26. August 2013 19:44
>>> To: user@synapse.apache.org
>>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>>
>>> Hi,
>>>
>>> On Aug 26, 2013, at 4:48 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>>
>>>> Hi Hiranja,
>>>>
>>>> No problem. Thanks for the support.
>>>>
>>>> Peter
>>>>
>>>> ===============================================================
>>>>
>>>> /*
>>>> * Filename: syslogMsgBuilder.java
>>>> * Author: P.Wright
>>>> * Date: 06.08.2013
>>>> * Description: Java class for Synapse class mediator.
>>>> * Class is embedded in Synapse, and called when a SYSLOG message is received.
>>>> *
>>>> * The SYSLOG message can contain 1-n syslog messages, each
>>>> * separated by a NL, and the length of each message defined
>>>> * at the start of it. This is called TCP-Framing "octet-counted".
>>>> * See: http://www.rsyslog.com/doc/omfwd.html
>>>> *
>>>> * Example input syslog message containing 3 messages
>>>> * 34 <1> This is dummy message number 1
>>>> * 34 <2> This is dummy message number 2
>>>> * 34 <3> This is dummy message number 3
>>>> *
>>>> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
>>>> * <?xml version='1.0' encoding='utf-8'?>
>>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>>> * <soapenv:Body>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload">
>>>> * 34 <1> This is dummy message number 1
>>>> * 34 <2> This is dummy message number 2
>>>> * 34 <3> This is dummy message number 3
>>>> * </text>
>>>> * </soapenv:Body>
>>>> * </soapenv:Envelope>
>>>> *
>>>> * Actions as follows:
>>>> * - Get the payload
>>>> * - Split the payload on NL
>>>> * - Remove the msglen at the start of each message
>>>> * - Replace any "<" with "<"
>>>> * - Replace the payload with an XML content with 3 messages (see below)
>>>> *
>>>> * Output message as follows:
>>>> * <?xml version='1.0' encoding='utf-8'?>
>>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>>> * <soapenv:Body>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
>>>> * </soapenv:Body>
>>>> * </soapenv:Envelope>
>>>> *
>>>> * Method Overview:
>>>> * - mediate ("main" method)
>>>> */
>>>> package com.sixtelekurs.classMediators;
>>>>
>>>> import java.util.Iterator;
>>>> import org.apache.axiom.om.OMElement;
>>>> import org.apache.axiom.om.util.AXIOMUtil;
>>>> import org.apache.axiom.soap.SOAPBody;
>>>> import org.apache.axiom.soap.SOAPEnvelope;
>>>> import org.apache.commons.logging.Log;
>>>> import org.apache.commons.logging.LogFactory;
>>>> import org.apache.synapse.MessageContext;
>>>> import org.apache.synapse.Mediator;
>>>>
>>>>
>>>> public class syslogMsgBuilder implements Mediator
>>>> {
>>>> private static String className = "syslogMsgBuilder";
>>>> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
>>>> private String _desc = null;
>>>> private String _uid = null;
>>>>
>>>> // ------------------------------------------------------------------------
>>>> public boolean mediate(MessageContext mc)
>>>> {
>>>> String logMsg = "===" + className + ".mediate called=============";
>>>> log.debug(logMsg);
>>>>
>>>> // Get the message content and process it
>>>> String syslogStr = mc.getEnvelope().getBody().getFirstElement().getText();
>>>
>>> I believe this line if the issue. The call to getText() on the OMSourcedElement probably causes it to read the entire input stream until end-of-stream is encountered. But end-of-stream does not occur until the sender closes its output stream at the remote end.
>>>
>>> I think a proper solution should get the InputStream from the OMSourcedElement and try to read from it until a known number of bytes have been read or a particular delimiter character is encountered. In a case like HTTP, the HTTP message usually have the content-length header or a special delimiter sequence (chunk delimiter) which allows the server to know how much data to read from the input stream without reading all the way to the end-of-stream. In case of TCP transport, there are no special headers to look at, so you must rely on the information available on the message payload it self.
>>>
>>> That's just my take on it. May be there's a better/easier way to solve this problem at Axiom level (perhaps a special method in the OMSourcedElement API?). May be Andreas can shed some light on the matter.
>>>
>>> Thanks,
>>> Hiranya
>>>
>>>> logMsg = "syslogStr='" + syslogStr + "'";
>>>> log.debug(logMsg);
>>>>
>>>> // Delete the existing payload
>>>> logMsg = "Deleting existing payload";
>>>> log.debug(logMsg);
>>>> SOAPBody soapBody = mc.getEnvelope().getBody();
>>>> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
>>>> {
>>>> itr.next();
>>>> itr.remove();
>>>> }
>>>>
>>>> // Split the strings into N syslog messages, convert each to
>>>> // an OMElement, and append to the XML-Body
>>>> int msgCnt = 0;
>>>> try
>>>> {
>>>> // Split the string on NL
>>>> String[] lines = syslogStr.split("\\n");
>>>> logMsg = "syslog.input.cnt(NL)=" + lines.length;
>>>> log.debug(logMsg);
>>>> for (int ii=0; ii < lines.length; ii++)
>>>> {
>>>> // Now split on space (once only) to ignore the msglen
>>>> String[] msg = lines[ii].split(" ", 2);
>>>> if (msg.length == 2)
>>>> {
>>>> // Replace all "<" with "<" for XML
>>>> String syslogMsg = msg[1].replaceAll("<", "<");
>>>> int jj = ii+1;
>>>> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
>>>> log.debug(logMsg);
>>>> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
>>>> xmlStr += syslogMsg;
>>>> xmlStr += "</text>";
>>>> msgCnt++;
>>>>
>>>> // Now add elem
>>>> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
>>>> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
>>>> log.debug(logMsg);
>>>> soapBody.addChild(xmlElem);
>>>> }
>>>> else
>>>> {
>>>> logMsg = "msg could not be split correctly [" + msg + "]";
>>>> log.error(logMsg);
>>>> }
>>>> }
>>>> }
>>>> catch (Exception e)
>>>> {
>>>> logMsg = "ERROR: " + e;
>>>> log.error(logMsg);
>>>> return(false);
>>>> }
>>>>
>>>> if (msgCnt == 0)
>>>> {
>>>> logMsg = "ERROR: no messages processed";
>>>> log.error(logMsg);
>>>> return(false);
>>>> }
>>>>
>>>> logMsg = "Made " + msgCnt + " messages";
>>>> log.info(logMsg);
>>>>
>>>> SOAPEnvelope envelope = mc.getEnvelope();
>>>> logMsg = "SOAP.Envelope: " + envelope;
>>>> log.debug(logMsg);
>>>>
>>>> return(true);
>>>> }
>>>>
>>>> // SET methods --------------------------------
>>>> public String getDescription()
>>>> {
>>>> String msg = className + ".desc";
>>>> return(msg);
>>>> }
>>>> public String getType()
>>>> {
>>>> String msg = className + ".type";
>>>> return(msg);
>>>> }
>>>> public int getTraceState()
>>>> {
>>>> return(0);
>>>> }
>>>>
>>>> // SET methods --------------------------------
>>>> public void setTraceState(int traceState)
>>>> {
>>>> traceState = 0;
>>>> }
>>>> public void setDescription(String desc)
>>>> {
>>>> _desc = desc;
>>>> }
>>>> public void setUid(String uid)
>>>> {
>>>> _uid = uid;
>>>> }
>>>> }
>>>>
>>>> -----Original Message-----
>>>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>>>> Sent: Freitag, 23. August 2013 08:30
>>>> To: user@synapse.apache.org
>>>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>>>
>>>> It sounds like something is trying to read from the input stream until the end of stream is encountered. This may be the plain text builder or your custom mediator. Can you share your custom mediator source code so we can take a look?
>>>>
>>>> Thanks,
>>>> Hiranya
>>>>
>>>> On Aug 22, 2013, at 8:27 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>>>
>>>>> Hi,
>>>>>
>>>>> I have my TCP Syslog --> JMS proxy up and running (see previous emails), but now have encountered another problem.
>>>>> Up till now I have simply tested by using netstat to send 1-n messages stored in a file to the TCP server.
>>>>> In this case, the messages are being received and processed correctly by synapse (sent as JMS
>>>>> messages to the JMS server). It seems that this is working correctly, because the connection to the
>>>>> synapse TCP server is being closed by netstat after it sends the messages.
>>>>>
>>>>> In my real world case however, we have 1-n clients writing syslog messages to a
>>>>> local syslog server. This syslog server has been configured to send the messages on
>>>>> to the (remote) synapse TCP server. In this case the messages are NOT being received
>>>>> and processed by synapse. Only after the syslog server closes it connection to the synapse
>>>>> TCP server, are the messages being received and processed by synapse.
>>>>>
>>>>> Any ideas what the problem could be?
>>>>> Attached again my synapse.xml config file.
>>>>>
>>>>> Thanks,
>>>>> Peter
>>>>>
>>>>> ===============================================
>>>>>
>>>>> <definitions xmlns="http://ws.apache.org/ns/synapse">
>>>>> <sequence name="fault">
>>>>> <makefault>
>>>>> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
>>>>> <reason value="Mediation failed."/>
>>>>> </makefault>
>>>>> <send/>
>>>>> </sequence>
>>>>> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
>>>>> <in>
>>>>> <log level="full"/>
>>>>> <send/>
>>>>> </in>
>>>>> <out>
>>>>> <send/>
>>>>> </out>
>>>>> </sequence>
>>>>>
>>>>> <proxy name="proxyTcp2Jms" transports="tcp">
>>>>> <target>
>>>>> <inSequence>
>>>>> <!-- Define TCP listener expects plain text (syslog) messages -->
>>>>> <property name="messageType" value="text/plain" scope="axis2"/>
>>>>> <property name="OUT_ONLY" value="true"/>
>>>>> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
>>>>> <log level="full"/>
>>>>>
>>>>> <!-- Split any multiple syslog messages for the iterator -->
>>>>> <!-- and return as XML with child messages -->
>>>>> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
>>>>> <log level="full"/>
>>>>> </class>
>>>>>
>>>>> <log level="full"/>
>>>>>
>>>>> <!-- Iterate over any multiple messages -->
>>>>> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
>>>>> <target>
>>>>> <sequence>
>>>>> <send>
>>>>> <endpoint>
>>>>> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
>>>>> </endpoint>
>>>>> </send>
>>>>> </sequence>
>>>>> </target>
>>>>> </iterate>
>>>>> </inSequence>
>>>>> <outSequence/>
>>>>> <faultSequence>
>>>>> <log level="full" category="ERROR" separator=","/>
>>>>> </faultSequence>
>>>>> </target>
>>>>> <parameter name="transport.tcp.port">6060</parameter>
>>>>> <parameter name="transport.tcp.contentType">text/plain</parameter>
>>>>> </proxy>
>>>>>
>>>>> </definitions>
>>>>>
>>>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>>>> Thank you.
>>>>
>>>> --
>>>> Hiranya Jayathilaka
>>>> Mayhem Lab/RACE Lab;
>>>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>>>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>>>> Blog: http://techfeast-hiranya.blogspot.com
>>>>
>>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>>> Thank you.
>>>
>>> --
>>> Hiranya Jayathilaka
>>> Mayhem Lab/RACE Lab;
>>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>>> Blog: http://techfeast-hiranya.blogspot.com
>>>
>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>> Thank you.
>>
>> --
>> Hiranya Jayathilaka
>> Mayhem Lab/RACE Lab;
>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>> Blog: http://techfeast-hiranya.blogspot.com
>>
>> The content of this e-mail is intended only for the confidential use of the person addressed.
>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>> Thank you.
>
> --
> Hiranya Jayathilaka
> Mayhem Lab/RACE Lab;
> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
> Blog: http://techfeast-hiranya.blogspot.com
>
> The content of this e-mail is intended only for the confidential use of the person addressed.
> If you are not the intended recipient, please notify the sender and delete this email immediately.
> Thank you.
RE: Synapse: TCP syslog to JMS proxy: messages only processed in
synapse after client TCP connection is closed
Posted by "Wright, Peter" <Pe...@six-group.com>.
Hi Hiranja,
I've done some more tests and have seen that my class can only work if
the client (in our case a syslog server configured to forward messages
to the synapse tcp listener) actually regularly opens/closes its connection
to the tcp listener.
Only when the connection is closed does my class mediator get all sent messages.
As long as the connection is open, my read blocks and my class mediator
cannot complete and pass the messages onto the synapse iterator mediator.
If I return before the connection is closed, all messages read up till
then will be sent on correctly to the iterator mediator, and will then
be forwarded as JMS messages as desired, but any remaining messages
sent by the syslog client will be lost.
The problem is that synapse works sequentially, but in this case an
asynchronous (callback?) type of mediator would be necessary to allow
the handling of each message as it is read.
I'm thinking of using the tcp listener and my class mediator, but
removing the iterator mediator. I could extend my class to directly
send each message as a JMS message to the messaging server (in java code).
My class would remain active as long as the tcp client keeps the connection open.
That way I could at least use the synapse multi threaded tcp listener
and not have to write that myself.
What do you think? Any other suggestions?
Thanks,
Peter
-----Original Message-----
From: Wright, Peter
Sent: Dienstag, 10. September 2013 16:25
To: 'user@synapse.apache.org'
Subject: RE: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
Hi Hiranja,
I'm still having a problem with finding "end of synapse message".
A synapse message can contain 1-n syslog messages.
Each syslog message is preceeded by its message length,
so I have no problem reading each message.
The problem is to know when to stop trying to read a possible
next message i.e. end of synapse message (end of frame?).
When I try to read one byte of the length of the next
(non existing) message, the read blocks.
Every message has a terminating newline character, which
is not necessary, as the message length is defined anyway.
If the newline were only at the end of frame (synapse message),
I could use that to know I should stop reading.
Who is adding the newlines? The syslog sender, or the TCP transport
module of synapse? Any chance of finding a solution here?
(end of frame delimiter?).
> TCP transport will inject all messages sent on the same connection as a single message into Synapse.
Does the TCP transport buffer the input syslog messages before concatenating them into the synapse message?
When are the messages sent? Buffer size reached? Time? Hopefully not only after the TCP connection is closed??
Is one synapse message equivalent to one TCP frame?
In our test setup, we have a syslog server configured to send all received messages
on to the synapse TCP listener.
Thanks
Peter
-----Original Message-----
From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
Sent: Donnerstag, 5. September 2013 19:59
To: user@synapse.apache.org
Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
You need to somehow figure out the number of messages to read from the input stream and read all those messages in your custom mediator. TCP transport will inject all messages sent on the same connection as a single message into Synapse.
If you want to debug your code, start synapse with the flag -xdebug and do a remote debug via port 8000.
Thanks,
Hiranya
On Sep 5, 2013, at 5:34 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
> Hi Hiranja,
>
> I've rewritten my class mediator to just read the message content according to the message lengths,
> and changed the log mediator settings as you suggested, but there still appears to be a problem.
>
> When a client writes 3 separate messages, opening and closing the TCP connection each time, all messages
> are received and processed correctly by synapse.
> However, when a client opens a TCP connection, writes 3 separate messages, and then closes the connection,
> only the first message is received and processed correctly by synapse.
>
> My class mediator processes the first message correctly, so it appears(?) to be processing and returning
> the message content correctly (without blocking), but there are no log entries at all for the following
> messages.
>
> I've attached my various config files and the source code of my class mediator.
>
> Thanks for your help,
> Peter
>
> ======================================
>
>
> synapse.log
> -----------
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,MediatorLog.java:97,To: , WSAction: urn:mediate, SOAPAction: urn:mediate, MessageID: urn:uuid:B3133E5C6DF9C641551378383293591, Direction: request
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:78,===syslogMsgBuilder.mediate called=============
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:161,getSyslogMsgs. start
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:171,getSyslogMsgs. Reading msg 1 =============
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:229,readMsgLen. start
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.0=1
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.1=9
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.2=4
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:260,readMsgLen: strMsglen = '194'
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:283,readMsgLen: intMsglen = 194
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:299,readMsgContent. start. readLen=194
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:342,readMsgContent: msgContent=[<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] ], nlFound=false
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:195,getSyslogMsgs: no NL found: stop
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:85,Found 1 message(s). Delete existing payload
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:105,syslogMsg.1='<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] '
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:113,xmlStr.1='<text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text>'
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:132,Made 1 message(s)
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:136,SOAP.Envelope: <?xml version='1.0' encoding='utf-8'?><soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"><soapenv:Body><text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text></soapenv:Body></soapenv:Envelope>
>
> synapse.xml
> -----------
> <definitions xmlns="http://ws.apache.org/ns/synapse">
> <sequence name="fault">
> <makefault>
> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
> <reason value="Mediation failed."/>
> </makefault>
> <send/>
> </sequence>
> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
> <in>
> <log level="simple"/>
> <send/>
> </in>
> <out>
> <send/>
> </out>
> </sequence>
>
> <proxy name="proxyTcp2Jms" transports="tcp">
> <target>
> <inSequence>
> <!-- Define TCP listener expects plain text (syslog) messages -->
> <property name="messageType" value="text/plain" scope="axis2"/>
> <property name="OUT_ONLY" value="true"/>
> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
> <log level="simple"/> <!-- simple,headers,full -->
>
> <!-- Split any multiple syslog messages for the iterator -->
> <!-- and return as XML with child messages -->
> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
> <!-- <log level="full"/> -->
> </class>
>
> <!-- <log level="full"/> -->
>
> <!-- Iterate over any multiple messages -->
> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
> <target>
> <sequence>
> <send>
> <!-- <log level="full"/> -->
> <endpoint>
> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
> </endpoint>
> </send>
> </sequence>
> </target>
> </iterate>
> </inSequence>
> <!-- OUT_ONLY=true
> <outSequence>
> <log level="simple" category="INFO" separator=","/>
> <send/>
> </outSequence>
> -->
> <faultSequence>
> <log level="simple" category="ERROR" separator=","/>
> </faultSequence>
> </target>
> <parameter name="transport.tcp.port">6060</parameter>
> <parameter name="transport.tcp.contentType">text/plain</parameter>
> <!-- <parameter name="transport.tcp.backlog">0</parameter> -->
> </proxy>
> </definitions>
>
>
> syslogMsgBuilder.java
> ---------------------
> /*
> * Filename: syslogMsgBuilder.java
> * Author: P.Wright
> * Date: 06.08.2013
> * Description: Java class for Synapse class mediator.
> * Class is embedded in Synapse, and called when a SYSLOG message is received.
> *
> * The SYSLOG message can contain 1-n syslog messages, each
> * separated by a NL, and the length of each message defined
> * at the start of it. This is called TCP-Framing "octet-counted".
> * See: http://www.rsyslog.com/doc/omfwd.html
> *
> * Example input syslog message containing 3 messages
> * 35 <1> This is dummy message number 1\n
> * 35 <2> This is dummy message number 2\n
> * 34 <3> This is dummy message number 3
> *
> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
> * <?xml version='1.0' encoding='utf-8'?>
> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
> * <soapenv:Body>
> * <text xmlns="http://ws.apache.org/commons/ns/payload">
> * 35 <1> This is dummy message number 1\n
> * 35 <2> This is dummy message number 2\n
> * 34 <3> This is dummy message number 3
> * </text>
> * </soapenv:Body>
> * </soapenv:Envelope>
> *
> * Actions as follows:
> * - Get the first element of the message body
> * - Get a reader to the content of the first element
> * - Read the msglen
> * - Read the message according to msglen
> * - If last character is NOT a '\n', stop reading messages
> * - If last character IS a '\n', keep going and read next msglen
> * - Replace any "<" with "<"
> * - Replace the payload with an XML content with 3 messages (see below)
> *
> * Output message as follows:
> * <?xml version='1.0' encoding='utf-8'?>
> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
> * <soapenv:Body>
> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
> * </soapenv:Body>
> * </soapenv:Envelope>
> *
> * Method Overview:
> * - mediate ("main" method)
> */
>
> package com.sixtelekurs.classMediators;
>
> import java.io.Reader;
> import java.util.Iterator;
> import java.util.ArrayList;
> import java.util.List;
> import org.apache.axiom.om.OMElement;
> import org.apache.axiom.om.util.AXIOMUtil;
> import org.apache.axiom.soap.SOAPBody;
> import org.apache.axiom.soap.SOAPEnvelope;
> import org.apache.commons.logging.Log;
> import org.apache.commons.logging.LogFactory;
> import org.apache.synapse.MessageContext;
> import org.apache.synapse.Mediator;
>
>
> public class syslogMsgBuilder implements Mediator
> {
> private static String className = "syslogMsgBuilder";
> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
> private String _desc = null;
> private String _uid = null;
>
> // ------------------------------------------------------------------------
> public boolean mediate(MessageContext mc)
> {
> String logMsg = "===" + className + ".mediate called=============";
> log.info(logMsg);
>
> // Get the message content as an array of strings
> List<String> listSyslogStr = getSyslogMsgs(mc.getEnvelope().getBody().getFirstElement());
>
> // Delete the existing payload
> logMsg = "Found " + listSyslogStr.size() + " message(s). Delete existing payload";
> log.info(logMsg);
> SOAPBody soapBody = mc.getEnvelope().getBody();
> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
> {
> itr.next();
> itr.remove();
> }
>
> // Traverse the string list, convert each to an XML element
> // and write into the SOAP body
> int msgCnt = 0;
> try
> {
> // Get the syslog msg
> for (String msg : listSyslogStr)
> {
> // Replace all "<" with "<" for XML
> String syslogMsg = msg.replaceAll("<", "<");
> int jj = msgCnt+1;
> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
> log.debug(logMsg);
> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
> xmlStr += syslogMsg;
> xmlStr += "</text>";
> msgCnt++;
>
> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
> log.info(logMsg);
> soapBody.addChild(xmlElem);
> }
> }
> catch (Exception e)
> {
> logMsg = "ERROR: " + e;
> log.error(logMsg);
> return(false);
> }
>
> if (msgCnt == 0)
> {
> logMsg = "ERROR: no messages processed";
> log.error(logMsg);
> return(false);
> }
>
> logMsg = "Made " + msgCnt + " message(s)";
> log.info(logMsg);
>
> SOAPEnvelope envelope = mc.getEnvelope();
> logMsg = "SOAP.Envelope: " + envelope;
> log.info(logMsg);
>
> return(true);
> }
>
> class ReturnValues
> {
> String msgString;
> boolean foundNewline;
> ReturnValues(final String msgString, boolean foundNewline)
> {
> this.msgString = msgString;
> this.foundNewline = foundNewline;
> }
> }
>
> // getSyslogMsgs: traverses the input message
> // and returns any syslog messages as an array
> // of strings in XML form
> private List<String> getSyslogMsgs(OMElement omElem)
> {
> String funcname = "getSyslogMsgs";
> String logMsg = "";
>
> logMsg = funcname + ". start";
> log.debug(logMsg);
>
> List<String> syslogMsgs = new ArrayList<String>();
> Reader reader = omElem.getTextAsStream(false); // pass cache=false
>
> int msgCnt = 0;
> boolean keepGoing = true;
> while (keepGoing)
> {
> logMsg = funcname + ". Reading msg " + (msgCnt+1) + " =============";
> log.info(logMsg);
>
> // Get the msglen
> int msgLen = readMsgLen(reader);
> if (msgLen <= 0)
> {
> keepGoing = false;
> }
> else
> {
> // Get the message
> ReturnValues retvals = readMsgContent(reader, msgLen);
> if (retvals.msgString.isEmpty())
> {
> keepGoing = false;
> }
> else
> {
> syslogMsgs.add(retvals.msgString);
>
> // Read any trailing NL
> if (!retvals.foundNewline)
> {
> logMsg = funcname + ": no NL found: stop";
> log.debug(logMsg);
> keepGoing = false;
> }
> }
> }
> msgCnt++;
> }
>
> // Close the reader
> /*********************
> try
> {
> reader.close();
> }
> catch (Exception e)
> {
> logMsg = funcname + ": reader.close failed. " + e;
> log.error(logMsg);
> }
> *********************/
> return(syslogMsgs);
> }
>
> // Get the msglen of the syslog message
> // Read characters till a space and convert to an integer
> // Return the int msglen
> private int readMsgLen(Reader reader)
> {
> String funcname = "readMsgLen";
> int intMsglen = 0;
> String logMsg = "";
> String strMsglen = "";
>
> logMsg = funcname + ". start";
> log.debug(logMsg);
>
> try
> {
> char[] cBuffer = new char[10];
> int cCnt=0;
> boolean keepGoing = true;
> while (keepGoing)
> {
> // Read one character at a time into the char array
> int rc = reader.read(cBuffer, cCnt, 1);
> if (rc <= 0)
> {
> keepGoing = false;
> }
> else if (cBuffer[cCnt] != ' ')
> {
> logMsg = "cBuffer." + cCnt + "=" + cBuffer[cCnt];
> log.debug(logMsg);
> cCnt++;
> }
> else
> {
> keepGoing = false;
> }
> }
> if (cCnt > 0)
> {
> // Convert the characters read into a string
> strMsglen = new String(cBuffer, 0, cCnt);
> logMsg = funcname + ": strMsglen = '" + strMsglen + "'";
> log.debug(logMsg);
> }
> }
> catch (Exception e)
> {
> logMsg = funcname + ": error reading msglen" + e;
> log.error(logMsg);
> }
>
> if (strMsglen != "")
> {
> // Convert the (string)msglen to an (int)msglen
> try
> {
> intMsglen = Integer.parseInt(strMsglen);
> }
> catch (Exception e)
> {
> logMsg = funcname + ": error converting string '" + strMsglen + "' to integer. " + e;
> log.error(logMsg);
> }
> }
> logMsg = funcname + ": intMsglen = " + intMsglen;
> log.debug(logMsg);
> return(intMsglen);
> }
>
> // readMsgContent --------------------------------------------
> private ReturnValues readMsgContent(Reader reader, int msgLen)
> {
> String funcname = "readMsgContent";
> String msgContent = "";
> boolean nlFound = false;
> String logMsg = "";
> int maxlen = 2048;
> char[] cBuffer = new char[maxlen];
> int readLen = (msgLen < (maxlen-1) ? msgLen:(maxlen-1));
>
> logMsg = funcname + ". start. readLen=" + readLen;
> log.debug(logMsg);
>
> if (msgLen > maxlen-1)
> {
> logMsg = "message too long for buffer (" + msgLen + "/" + (maxlen-1);
> log.error(logMsg);
> ReturnValues retvals = new ReturnValues("", false);
> return(retvals);
> }
>
> try
> {
> int rc = reader.read(cBuffer, 0, readLen);
> if (rc <= 0)
> {
> logMsg = funcname + ": error reading msg: " + rc;
> log.error(logMsg);
> }
> else
> {
> if (rc < readLen)
> {
> logMsg = "read: IST=" + rc + ", SOLL=" + readLen;
> log.warn(logMsg);
> }
>
> // Ignore any trailing NL
> if (cBuffer[rc-1] == '\n')
> {
> nlFound = true;
> logMsg = "NL read at end of message - ignore it";
> log.debug(logMsg);
> rc--;
> }
> msgContent = new String(cBuffer, 0, rc);
> }
> }
> catch (Exception e)
> {
> logMsg = funcname + ": exception reading msg" + e;
> log.error(logMsg);
> }
> logMsg = funcname + ": msgContent=[" + msgContent + "], nlFound=" + nlFound;
> log.debug(logMsg);
>
> ReturnValues retvals = new ReturnValues(msgContent, nlFound);
> return(retvals);
> }
>
> // SET methods --------------------------------
> public String getDescription()
> {
> String msg = className + ".desc";
> return(msg);
> }
> public String getType()
> {
> String msg = className + ".type";
> return(msg);
> }
> public int getTraceState()
> {
> return(0);
> }
>
> // SET methods --------------------------------
> public void setTraceState(int traceState)
> {
> traceState = 0;
> }
> public void setDescription(String desc)
> {
> _desc = desc;
> }
> public void setUid(String uid)
> {
> _uid = uid;
> }
> }
>
> -----Original Message-----
> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
> Sent: Freitag, 30. August 2013 19:11
> To: user@synapse.apache.org
> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>
>
> On Aug 30, 2013, at 5:40 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>
>> Hi Hiranja,
>>
>> I've done a series of tests that appear to indicate the problem
>> is in the TCP transport.
>
> It's not really a problem in the tcp transport. The issue is exactly what I mentioned in my previous mail. That is, Synapse tries to read the input stream all the way to the end-of-stream, which causes it to get blocked. In addition to your custom mediator, other mediators like <lov level="full"/> also cause the same behavior. Basically any mediator that accesses the full message payload can cause this problem.
>
> The way I see it, you have only three options at this point:
>
> 1. Change your class mediator to not read the message payload to the end-of-stream. But you still have to get rid of any other mediators that may cause the message to be fully serialized (e.g. <log level="full"/>). Therefore this may not be very practical.
> 2. Write a custom message builder instead of a class mediator. This is probably the most architecturally sound approach. However, you must make sure that the builder doesn't read the input stream all the way to the end-of-stream. That is, it should read the message up to a known delimiter, or read a known number of bytes. Otherwise you will have the same problem again.
> 3. Use HTTP to receive the messages (HTTP protocol handles this type of issues using the Content-Length header or the chunked encoding system).
>
>> Here are the steps I did and the results I saw:
>> Important: "still the same" means:
>> - while the tcp connection is open, there is no reaction from synapse (no log messages etc)
>> - only AFTER the connection to the TCP listener has been closed do log messages appear and the syslog messages are processed
>>
>> 1. Rewrote my class mediator to use BufferedReader.readLine() -> still the same
>> 2. Rewrote my class mediator to simply replace the payload with fixed messages WITHOUT even reading the existing payload -> still the same
>> 3. Commented out my class mediator in synapse.xml -> still the same
>
> All the above 3 are mainly due to the other serializing mediators in the configuration (e.g. log). Also FYI, changing to the BufferedRead.readLine() doesn't make any difference. That is also a blocking I/O call.
>
>> 4. Added transport.tcp.port to the axis2.xml (as well as being in proxyTcp2Jms in synapse.xml) -> error at synapse startup "Error while starting the TCP endpoint. Address already in use"
>> 5. As above, but removed the port from proxyTcp2Jms in synapse.xml -> error at synapse startup "Service doesn't have configuration information for transport tcp"
>> 6. Defined transport.tcp.port in both files, but with different port numbers (6060 in synapse.xml, 6061 in axis2.xml) -> synapse starts up OK (with two logfile entries):
>> - TCPServer.java:76,TCP server started on port : 6061
>> - TCPServer.java:76,TCP server started on port : 6060
>> -> when I send a syslog message to port 6060 -> still the same
>> -> when I send a syslog message to port 6061 -> error written to logfile immediately: AxisEngine.java:219,The service cannot be found for the endpoint reference (EPR)
>> It appears that the TCPserver configured in synapse.xml is blocking until the client closes the connection,
>> and the TCPListener configured in axis2.xml is not blocking, but cannot find the proxy definition.
>
> That's not how it works. Messages sent to port 6061 must be dispatched by looking at the first element of the SOAP body. In your case it's just a generic <text/> element. Therefore Synapse fails to dispatch the message to any proxy service. This happens before any of the mediators are invoked. Therefore Synapse doesn't get to access the full message payload in this case. Hence it fails early in the pipeline, before Synapse gets blocked on I/O.
>
> Thanks,
> Hiranya
>
>>
>> The tcp jar file I'm using is this one:
>> http://maven.wso2.org/nexus/content/groups/wso2-public/org/apache/axis2/axis2-transport-tcp/1.1.0-SNAPSHOT/axis2-transport-tcp-1.1.0-SNAPSHOT.jar
>> but today also tried this one but saw no difference:
>> http://dist.wso2.org/maven2//org/apache/axis2/axis2-transport-tcp/1.1.0-wso2v5/axis2-transport-tcp-1.1.0-wso2v5.jar
>>
>> Thanks,
>> Peter
>>
>> -----Original Message-----
>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>> Sent: Montag, 26. August 2013 19:44
>> To: user@synapse.apache.org
>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>
>> Hi,
>>
>> On Aug 26, 2013, at 4:48 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>
>>> Hi Hiranja,
>>>
>>> No problem. Thanks for the support.
>>>
>>> Peter
>>>
>>> ===============================================================
>>>
>>> /*
>>> * Filename: syslogMsgBuilder.java
>>> * Author: P.Wright
>>> * Date: 06.08.2013
>>> * Description: Java class for Synapse class mediator.
>>> * Class is embedded in Synapse, and called when a SYSLOG message is received.
>>> *
>>> * The SYSLOG message can contain 1-n syslog messages, each
>>> * separated by a NL, and the length of each message defined
>>> * at the start of it. This is called TCP-Framing "octet-counted".
>>> * See: http://www.rsyslog.com/doc/omfwd.html
>>> *
>>> * Example input syslog message containing 3 messages
>>> * 34 <1> This is dummy message number 1
>>> * 34 <2> This is dummy message number 2
>>> * 34 <3> This is dummy message number 3
>>> *
>>> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
>>> * <?xml version='1.0' encoding='utf-8'?>
>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>> * <soapenv:Body>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload">
>>> * 34 <1> This is dummy message number 1
>>> * 34 <2> This is dummy message number 2
>>> * 34 <3> This is dummy message number 3
>>> * </text>
>>> * </soapenv:Body>
>>> * </soapenv:Envelope>
>>> *
>>> * Actions as follows:
>>> * - Get the payload
>>> * - Split the payload on NL
>>> * - Remove the msglen at the start of each message
>>> * - Replace any "<" with "<"
>>> * - Replace the payload with an XML content with 3 messages (see below)
>>> *
>>> * Output message as follows:
>>> * <?xml version='1.0' encoding='utf-8'?>
>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>> * <soapenv:Body>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
>>> * </soapenv:Body>
>>> * </soapenv:Envelope>
>>> *
>>> * Method Overview:
>>> * - mediate ("main" method)
>>> */
>>> package com.sixtelekurs.classMediators;
>>>
>>> import java.util.Iterator;
>>> import org.apache.axiom.om.OMElement;
>>> import org.apache.axiom.om.util.AXIOMUtil;
>>> import org.apache.axiom.soap.SOAPBody;
>>> import org.apache.axiom.soap.SOAPEnvelope;
>>> import org.apache.commons.logging.Log;
>>> import org.apache.commons.logging.LogFactory;
>>> import org.apache.synapse.MessageContext;
>>> import org.apache.synapse.Mediator;
>>>
>>>
>>> public class syslogMsgBuilder implements Mediator
>>> {
>>> private static String className = "syslogMsgBuilder";
>>> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
>>> private String _desc = null;
>>> private String _uid = null;
>>>
>>> // ------------------------------------------------------------------------
>>> public boolean mediate(MessageContext mc)
>>> {
>>> String logMsg = "===" + className + ".mediate called=============";
>>> log.debug(logMsg);
>>>
>>> // Get the message content and process it
>>> String syslogStr = mc.getEnvelope().getBody().getFirstElement().getText();
>>
>> I believe this line if the issue. The call to getText() on the OMSourcedElement probably causes it to read the entire input stream until end-of-stream is encountered. But end-of-stream does not occur until the sender closes its output stream at the remote end.
>>
>> I think a proper solution should get the InputStream from the OMSourcedElement and try to read from it until a known number of bytes have been read or a particular delimiter character is encountered. In a case like HTTP, the HTTP message usually have the content-length header or a special delimiter sequence (chunk delimiter) which allows the server to know how much data to read from the input stream without reading all the way to the end-of-stream. In case of TCP transport, there are no special headers to look at, so you must rely on the information available on the message payload it self.
>>
>> That's just my take on it. May be there's a better/easier way to solve this problem at Axiom level (perhaps a special method in the OMSourcedElement API?). May be Andreas can shed some light on the matter.
>>
>> Thanks,
>> Hiranya
>>
>>> logMsg = "syslogStr='" + syslogStr + "'";
>>> log.debug(logMsg);
>>>
>>> // Delete the existing payload
>>> logMsg = "Deleting existing payload";
>>> log.debug(logMsg);
>>> SOAPBody soapBody = mc.getEnvelope().getBody();
>>> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
>>> {
>>> itr.next();
>>> itr.remove();
>>> }
>>>
>>> // Split the strings into N syslog messages, convert each to
>>> // an OMElement, and append to the XML-Body
>>> int msgCnt = 0;
>>> try
>>> {
>>> // Split the string on NL
>>> String[] lines = syslogStr.split("\\n");
>>> logMsg = "syslog.input.cnt(NL)=" + lines.length;
>>> log.debug(logMsg);
>>> for (int ii=0; ii < lines.length; ii++)
>>> {
>>> // Now split on space (once only) to ignore the msglen
>>> String[] msg = lines[ii].split(" ", 2);
>>> if (msg.length == 2)
>>> {
>>> // Replace all "<" with "<" for XML
>>> String syslogMsg = msg[1].replaceAll("<", "<");
>>> int jj = ii+1;
>>> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
>>> log.debug(logMsg);
>>> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
>>> xmlStr += syslogMsg;
>>> xmlStr += "</text>";
>>> msgCnt++;
>>>
>>> // Now add elem
>>> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
>>> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
>>> log.debug(logMsg);
>>> soapBody.addChild(xmlElem);
>>> }
>>> else
>>> {
>>> logMsg = "msg could not be split correctly [" + msg + "]";
>>> log.error(logMsg);
>>> }
>>> }
>>> }
>>> catch (Exception e)
>>> {
>>> logMsg = "ERROR: " + e;
>>> log.error(logMsg);
>>> return(false);
>>> }
>>>
>>> if (msgCnt == 0)
>>> {
>>> logMsg = "ERROR: no messages processed";
>>> log.error(logMsg);
>>> return(false);
>>> }
>>>
>>> logMsg = "Made " + msgCnt + " messages";
>>> log.info(logMsg);
>>>
>>> SOAPEnvelope envelope = mc.getEnvelope();
>>> logMsg = "SOAP.Envelope: " + envelope;
>>> log.debug(logMsg);
>>>
>>> return(true);
>>> }
>>>
>>> // SET methods --------------------------------
>>> public String getDescription()
>>> {
>>> String msg = className + ".desc";
>>> return(msg);
>>> }
>>> public String getType()
>>> {
>>> String msg = className + ".type";
>>> return(msg);
>>> }
>>> public int getTraceState()
>>> {
>>> return(0);
>>> }
>>>
>>> // SET methods --------------------------------
>>> public void setTraceState(int traceState)
>>> {
>>> traceState = 0;
>>> }
>>> public void setDescription(String desc)
>>> {
>>> _desc = desc;
>>> }
>>> public void setUid(String uid)
>>> {
>>> _uid = uid;
>>> }
>>> }
>>>
>>> -----Original Message-----
>>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>>> Sent: Freitag, 23. August 2013 08:30
>>> To: user@synapse.apache.org
>>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>>
>>> It sounds like something is trying to read from the input stream until the end of stream is encountered. This may be the plain text builder or your custom mediator. Can you share your custom mediator source code so we can take a look?
>>>
>>> Thanks,
>>> Hiranya
>>>
>>> On Aug 22, 2013, at 8:27 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>>
>>>> Hi,
>>>>
>>>> I have my TCP Syslog --> JMS proxy up and running (see previous emails), but now have encountered another problem.
>>>> Up till now I have simply tested by using netstat to send 1-n messages stored in a file to the TCP server.
>>>> In this case, the messages are being received and processed correctly by synapse (sent as JMS
>>>> messages to the JMS server). It seems that this is working correctly, because the connection to the
>>>> synapse TCP server is being closed by netstat after it sends the messages.
>>>>
>>>> In my real world case however, we have 1-n clients writing syslog messages to a
>>>> local syslog server. This syslog server has been configured to send the messages on
>>>> to the (remote) synapse TCP server. In this case the messages are NOT being received
>>>> and processed by synapse. Only after the syslog server closes it connection to the synapse
>>>> TCP server, are the messages being received and processed by synapse.
>>>>
>>>> Any ideas what the problem could be?
>>>> Attached again my synapse.xml config file.
>>>>
>>>> Thanks,
>>>> Peter
>>>>
>>>> ===============================================
>>>>
>>>> <definitions xmlns="http://ws.apache.org/ns/synapse">
>>>> <sequence name="fault">
>>>> <makefault>
>>>> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
>>>> <reason value="Mediation failed."/>
>>>> </makefault>
>>>> <send/>
>>>> </sequence>
>>>> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
>>>> <in>
>>>> <log level="full"/>
>>>> <send/>
>>>> </in>
>>>> <out>
>>>> <send/>
>>>> </out>
>>>> </sequence>
>>>>
>>>> <proxy name="proxyTcp2Jms" transports="tcp">
>>>> <target>
>>>> <inSequence>
>>>> <!-- Define TCP listener expects plain text (syslog) messages -->
>>>> <property name="messageType" value="text/plain" scope="axis2"/>
>>>> <property name="OUT_ONLY" value="true"/>
>>>> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
>>>> <log level="full"/>
>>>>
>>>> <!-- Split any multiple syslog messages for the iterator -->
>>>> <!-- and return as XML with child messages -->
>>>> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
>>>> <log level="full"/>
>>>> </class>
>>>>
>>>> <log level="full"/>
>>>>
>>>> <!-- Iterate over any multiple messages -->
>>>> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
>>>> <target>
>>>> <sequence>
>>>> <send>
>>>> <endpoint>
>>>> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
>>>> </endpoint>
>>>> </send>
>>>> </sequence>
>>>> </target>
>>>> </iterate>
>>>> </inSequence>
>>>> <outSequence/>
>>>> <faultSequence>
>>>> <log level="full" category="ERROR" separator=","/>
>>>> </faultSequence>
>>>> </target>
>>>> <parameter name="transport.tcp.port">6060</parameter>
>>>> <parameter name="transport.tcp.contentType">text/plain</parameter>
>>>> </proxy>
>>>>
>>>> </definitions>
>>>>
>>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>>> Thank you.
>>>
>>> --
>>> Hiranya Jayathilaka
>>> Mayhem Lab/RACE Lab;
>>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>>> Blog: http://techfeast-hiranya.blogspot.com
>>>
>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>> Thank you.
>>
>> --
>> Hiranya Jayathilaka
>> Mayhem Lab/RACE Lab;
>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>> Blog: http://techfeast-hiranya.blogspot.com
>>
>> The content of this e-mail is intended only for the confidential use of the person addressed.
>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>> Thank you.
>
> --
> Hiranya Jayathilaka
> Mayhem Lab/RACE Lab;
> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
> Blog: http://techfeast-hiranya.blogspot.com
>
> The content of this e-mail is intended only for the confidential use of the person addressed.
> If you are not the intended recipient, please notify the sender and delete this email immediately.
> Thank you.
--
Hiranya Jayathilaka
Mayhem Lab/RACE Lab;
Dept. of Computer Science, UCSB; http://cs.ucsb.edu
E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
Blog: http://techfeast-hiranya.blogspot.com
The content of this e-mail is intended only for the confidential use of the person addressed.
If you are not the intended recipient, please notify the sender and delete this email immediately.
Thank you.
Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
Posted by Hiranya Jayathilaka <hi...@gmail.com>.
On Sep 10, 2013, at 7:25 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
> Hi Hiranja,
>
> I'm still having a problem with finding "end of synapse message".
>
> A synapse message can contain 1-n syslog messages.
> Each syslog message is preceeded by its message length,
> so I have no problem reading each message.
>
> The problem is to know when to stop trying to read a possible
> next message i.e. end of synapse message (end of frame?).
> When I try to read one byte of the length of the next
> (non existing) message, the read blocks.
Your message should specify the exact total number of bytes at the beginning of the message. This is how most protocols work, including HTTP (using the Content-Length header). Then your mediator can read that amount of bytes from the input stream, turn the whole thing into a string or a byte array, and split it into separate syslog messages.
The other option is to NOT send multiple syslog messages as a single frame. Since your syslog entries seem to specify their individual lengths, your mediator should be able to parse them properly without getting blocked.
Thanks,
Hiranya
>
> Every message has a terminating newline character, which
> is not necessary, as the message length is defined anyway.
> If the newline were only at the end of frame (synapse message),
> I could use that to know I should stop reading.
>
> Who is adding the newlines? The syslog sender, or the TCP transport
> module of synapse? Any chance of finding a solution here?
> (end of frame delimiter?).
>
>> TCP transport will inject all messages sent on the same connection as a single message into Synapse.
> Does the TCP transport buffer the input syslog messages before concatenating them into the synapse message?
> When are the messages sent? Buffer size reached? Time? Hopefully not only after the TCP connection is closed??
> Is one synapse message equivalent to one TCP frame?
>
> In our test setup, we have a syslog server configured to send all received messages
> on to the synapse TCP listener.
>
> Thanks
> Peter
>
> -----Original Message-----
> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
> Sent: Donnerstag, 5. September 2013 19:59
> To: user@synapse.apache.org
> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>
> You need to somehow figure out the number of messages to read from the input stream and read all those messages in your custom mediator. TCP transport will inject all messages sent on the same connection as a single message into Synapse.
>
> If you want to debug your code, start synapse with the flag -xdebug and do a remote debug via port 8000.
>
> Thanks,
> Hiranya
>
> On Sep 5, 2013, at 5:34 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>
>> Hi Hiranja,
>>
>> I've rewritten my class mediator to just read the message content according to the message lengths,
>> and changed the log mediator settings as you suggested, but there still appears to be a problem.
>>
>> When a client writes 3 separate messages, opening and closing the TCP connection each time, all messages
>> are received and processed correctly by synapse.
>> However, when a client opens a TCP connection, writes 3 separate messages, and then closes the connection,
>> only the first message is received and processed correctly by synapse.
>>
>> My class mediator processes the first message correctly, so it appears(?) to be processing and returning
>> the message content correctly (without blocking), but there are no log entries at all for the following
>> messages.
>>
>> I've attached my various config files and the source code of my class mediator.
>>
>> Thanks for your help,
>> Peter
>>
>> ======================================
>>
>>
>> synapse.log
>> -----------
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,MediatorLog.java:97,To: , WSAction: urn:mediate, SOAPAction: urn:mediate, MessageID: urn:uuid:B3133E5C6DF9C641551378383293591, Direction: request
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:78,===syslogMsgBuilder.mediate called=============
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:161,getSyslogMsgs. start
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:171,getSyslogMsgs. Reading msg 1 =============
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:229,readMsgLen. start
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.0=1
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.1=9
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.2=4
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:260,readMsgLen: strMsglen = '194'
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:283,readMsgLen: intMsglen = 194
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:299,readMsgContent. start. readLen=194
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:342,readMsgContent: msgContent=[<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] ], nlFound=false
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:195,getSyslogMsgs: no NL found: stop
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:85,Found 1 message(s). Delete existing payload
>> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:105,syslogMsg.1='<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] '
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:113,xmlStr.1='<text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text>'
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:132,Made 1 message(s)
>> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:136,SOAP.Envelope: <?xml version='1.0' encoding='utf-8'?><soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"><soapenv:Body><text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text></soapenv:Body></soapenv:Envelope>
>>
>> synapse.xml
>> -----------
>> <definitions xmlns="http://ws.apache.org/ns/synapse">
>> <sequence name="fault">
>> <makefault>
>> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
>> <reason value="Mediation failed."/>
>> </makefault>
>> <send/>
>> </sequence>
>> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
>> <in>
>> <log level="simple"/>
>> <send/>
>> </in>
>> <out>
>> <send/>
>> </out>
>> </sequence>
>>
>> <proxy name="proxyTcp2Jms" transports="tcp">
>> <target>
>> <inSequence>
>> <!-- Define TCP listener expects plain text (syslog) messages -->
>> <property name="messageType" value="text/plain" scope="axis2"/>
>> <property name="OUT_ONLY" value="true"/>
>> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
>> <log level="simple"/> <!-- simple,headers,full -->
>>
>> <!-- Split any multiple syslog messages for the iterator -->
>> <!-- and return as XML with child messages -->
>> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
>> <!-- <log level="full"/> -->
>> </class>
>>
>> <!-- <log level="full"/> -->
>>
>> <!-- Iterate over any multiple messages -->
>> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
>> <target>
>> <sequence>
>> <send>
>> <!-- <log level="full"/> -->
>> <endpoint>
>> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
>> </endpoint>
>> </send>
>> </sequence>
>> </target>
>> </iterate>
>> </inSequence>
>> <!-- OUT_ONLY=true
>> <outSequence>
>> <log level="simple" category="INFO" separator=","/>
>> <send/>
>> </outSequence>
>> -->
>> <faultSequence>
>> <log level="simple" category="ERROR" separator=","/>
>> </faultSequence>
>> </target>
>> <parameter name="transport.tcp.port">6060</parameter>
>> <parameter name="transport.tcp.contentType">text/plain</parameter>
>> <!-- <parameter name="transport.tcp.backlog">0</parameter> -->
>> </proxy>
>> </definitions>
>>
>>
>> syslogMsgBuilder.java
>> ---------------------
>> /*
>> * Filename: syslogMsgBuilder.java
>> * Author: P.Wright
>> * Date: 06.08.2013
>> * Description: Java class for Synapse class mediator.
>> * Class is embedded in Synapse, and called when a SYSLOG message is received.
>> *
>> * The SYSLOG message can contain 1-n syslog messages, each
>> * separated by a NL, and the length of each message defined
>> * at the start of it. This is called TCP-Framing "octet-counted".
>> * See: http://www.rsyslog.com/doc/omfwd.html
>> *
>> * Example input syslog message containing 3 messages
>> * 35 <1> This is dummy message number 1\n
>> * 35 <2> This is dummy message number 2\n
>> * 34 <3> This is dummy message number 3
>> *
>> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
>> * <?xml version='1.0' encoding='utf-8'?>
>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>> * <soapenv:Body>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload">
>> * 35 <1> This is dummy message number 1\n
>> * 35 <2> This is dummy message number 2\n
>> * 34 <3> This is dummy message number 3
>> * </text>
>> * </soapenv:Body>
>> * </soapenv:Envelope>
>> *
>> * Actions as follows:
>> * - Get the first element of the message body
>> * - Get a reader to the content of the first element
>> * - Read the msglen
>> * - Read the message according to msglen
>> * - If last character is NOT a '\n', stop reading messages
>> * - If last character IS a '\n', keep going and read next msglen
>> * - Replace any "<" with "<"
>> * - Replace the payload with an XML content with 3 messages (see below)
>> *
>> * Output message as follows:
>> * <?xml version='1.0' encoding='utf-8'?>
>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>> * <soapenv:Body>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
>> * </soapenv:Body>
>> * </soapenv:Envelope>
>> *
>> * Method Overview:
>> * - mediate ("main" method)
>> */
>>
>> package com.sixtelekurs.classMediators;
>>
>> import java.io.Reader;
>> import java.util.Iterator;
>> import java.util.ArrayList;
>> import java.util.List;
>> import org.apache.axiom.om.OMElement;
>> import org.apache.axiom.om.util.AXIOMUtil;
>> import org.apache.axiom.soap.SOAPBody;
>> import org.apache.axiom.soap.SOAPEnvelope;
>> import org.apache.commons.logging.Log;
>> import org.apache.commons.logging.LogFactory;
>> import org.apache.synapse.MessageContext;
>> import org.apache.synapse.Mediator;
>>
>>
>> public class syslogMsgBuilder implements Mediator
>> {
>> private static String className = "syslogMsgBuilder";
>> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
>> private String _desc = null;
>> private String _uid = null;
>>
>> // ------------------------------------------------------------------------
>> public boolean mediate(MessageContext mc)
>> {
>> String logMsg = "===" + className + ".mediate called=============";
>> log.info(logMsg);
>>
>> // Get the message content as an array of strings
>> List<String> listSyslogStr = getSyslogMsgs(mc.getEnvelope().getBody().getFirstElement());
>>
>> // Delete the existing payload
>> logMsg = "Found " + listSyslogStr.size() + " message(s). Delete existing payload";
>> log.info(logMsg);
>> SOAPBody soapBody = mc.getEnvelope().getBody();
>> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
>> {
>> itr.next();
>> itr.remove();
>> }
>>
>> // Traverse the string list, convert each to an XML element
>> // and write into the SOAP body
>> int msgCnt = 0;
>> try
>> {
>> // Get the syslog msg
>> for (String msg : listSyslogStr)
>> {
>> // Replace all "<" with "<" for XML
>> String syslogMsg = msg.replaceAll("<", "<");
>> int jj = msgCnt+1;
>> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
>> log.debug(logMsg);
>> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
>> xmlStr += syslogMsg;
>> xmlStr += "</text>";
>> msgCnt++;
>>
>> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
>> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
>> log.info(logMsg);
>> soapBody.addChild(xmlElem);
>> }
>> }
>> catch (Exception e)
>> {
>> logMsg = "ERROR: " + e;
>> log.error(logMsg);
>> return(false);
>> }
>>
>> if (msgCnt == 0)
>> {
>> logMsg = "ERROR: no messages processed";
>> log.error(logMsg);
>> return(false);
>> }
>>
>> logMsg = "Made " + msgCnt + " message(s)";
>> log.info(logMsg);
>>
>> SOAPEnvelope envelope = mc.getEnvelope();
>> logMsg = "SOAP.Envelope: " + envelope;
>> log.info(logMsg);
>>
>> return(true);
>> }
>>
>> class ReturnValues
>> {
>> String msgString;
>> boolean foundNewline;
>> ReturnValues(final String msgString, boolean foundNewline)
>> {
>> this.msgString = msgString;
>> this.foundNewline = foundNewline;
>> }
>> }
>>
>> // getSyslogMsgs: traverses the input message
>> // and returns any syslog messages as an array
>> // of strings in XML form
>> private List<String> getSyslogMsgs(OMElement omElem)
>> {
>> String funcname = "getSyslogMsgs";
>> String logMsg = "";
>>
>> logMsg = funcname + ". start";
>> log.debug(logMsg);
>>
>> List<String> syslogMsgs = new ArrayList<String>();
>> Reader reader = omElem.getTextAsStream(false); // pass cache=false
>>
>> int msgCnt = 0;
>> boolean keepGoing = true;
>> while (keepGoing)
>> {
>> logMsg = funcname + ". Reading msg " + (msgCnt+1) + " =============";
>> log.info(logMsg);
>>
>> // Get the msglen
>> int msgLen = readMsgLen(reader);
>> if (msgLen <= 0)
>> {
>> keepGoing = false;
>> }
>> else
>> {
>> // Get the message
>> ReturnValues retvals = readMsgContent(reader, msgLen);
>> if (retvals.msgString.isEmpty())
>> {
>> keepGoing = false;
>> }
>> else
>> {
>> syslogMsgs.add(retvals.msgString);
>>
>> // Read any trailing NL
>> if (!retvals.foundNewline)
>> {
>> logMsg = funcname + ": no NL found: stop";
>> log.debug(logMsg);
>> keepGoing = false;
>> }
>> }
>> }
>> msgCnt++;
>> }
>>
>> // Close the reader
>> /*********************
>> try
>> {
>> reader.close();
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": reader.close failed. " + e;
>> log.error(logMsg);
>> }
>> *********************/
>> return(syslogMsgs);
>> }
>>
>> // Get the msglen of the syslog message
>> // Read characters till a space and convert to an integer
>> // Return the int msglen
>> private int readMsgLen(Reader reader)
>> {
>> String funcname = "readMsgLen";
>> int intMsglen = 0;
>> String logMsg = "";
>> String strMsglen = "";
>>
>> logMsg = funcname + ". start";
>> log.debug(logMsg);
>>
>> try
>> {
>> char[] cBuffer = new char[10];
>> int cCnt=0;
>> boolean keepGoing = true;
>> while (keepGoing)
>> {
>> // Read one character at a time into the char array
>> int rc = reader.read(cBuffer, cCnt, 1);
>> if (rc <= 0)
>> {
>> keepGoing = false;
>> }
>> else if (cBuffer[cCnt] != ' ')
>> {
>> logMsg = "cBuffer." + cCnt + "=" + cBuffer[cCnt];
>> log.debug(logMsg);
>> cCnt++;
>> }
>> else
>> {
>> keepGoing = false;
>> }
>> }
>> if (cCnt > 0)
>> {
>> // Convert the characters read into a string
>> strMsglen = new String(cBuffer, 0, cCnt);
>> logMsg = funcname + ": strMsglen = '" + strMsglen + "'";
>> log.debug(logMsg);
>> }
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": error reading msglen" + e;
>> log.error(logMsg);
>> }
>>
>> if (strMsglen != "")
>> {
>> // Convert the (string)msglen to an (int)msglen
>> try
>> {
>> intMsglen = Integer.parseInt(strMsglen);
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": error converting string '" + strMsglen + "' to integer. " + e;
>> log.error(logMsg);
>> }
>> }
>> logMsg = funcname + ": intMsglen = " + intMsglen;
>> log.debug(logMsg);
>> return(intMsglen);
>> }
>>
>> // readMsgContent --------------------------------------------
>> private ReturnValues readMsgContent(Reader reader, int msgLen)
>> {
>> String funcname = "readMsgContent";
>> String msgContent = "";
>> boolean nlFound = false;
>> String logMsg = "";
>> int maxlen = 2048;
>> char[] cBuffer = new char[maxlen];
>> int readLen = (msgLen < (maxlen-1) ? msgLen:(maxlen-1));
>>
>> logMsg = funcname + ". start. readLen=" + readLen;
>> log.debug(logMsg);
>>
>> if (msgLen > maxlen-1)
>> {
>> logMsg = "message too long for buffer (" + msgLen + "/" + (maxlen-1);
>> log.error(logMsg);
>> ReturnValues retvals = new ReturnValues("", false);
>> return(retvals);
>> }
>>
>> try
>> {
>> int rc = reader.read(cBuffer, 0, readLen);
>> if (rc <= 0)
>> {
>> logMsg = funcname + ": error reading msg: " + rc;
>> log.error(logMsg);
>> }
>> else
>> {
>> if (rc < readLen)
>> {
>> logMsg = "read: IST=" + rc + ", SOLL=" + readLen;
>> log.warn(logMsg);
>> }
>>
>> // Ignore any trailing NL
>> if (cBuffer[rc-1] == '\n')
>> {
>> nlFound = true;
>> logMsg = "NL read at end of message - ignore it";
>> log.debug(logMsg);
>> rc--;
>> }
>> msgContent = new String(cBuffer, 0, rc);
>> }
>> }
>> catch (Exception e)
>> {
>> logMsg = funcname + ": exception reading msg" + e;
>> log.error(logMsg);
>> }
>> logMsg = funcname + ": msgContent=[" + msgContent + "], nlFound=" + nlFound;
>> log.debug(logMsg);
>>
>> ReturnValues retvals = new ReturnValues(msgContent, nlFound);
>> return(retvals);
>> }
>>
>> // SET methods --------------------------------
>> public String getDescription()
>> {
>> String msg = className + ".desc";
>> return(msg);
>> }
>> public String getType()
>> {
>> String msg = className + ".type";
>> return(msg);
>> }
>> public int getTraceState()
>> {
>> return(0);
>> }
>>
>> // SET methods --------------------------------
>> public void setTraceState(int traceState)
>> {
>> traceState = 0;
>> }
>> public void setDescription(String desc)
>> {
>> _desc = desc;
>> }
>> public void setUid(String uid)
>> {
>> _uid = uid;
>> }
>> }
>>
>> -----Original Message-----
>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>> Sent: Freitag, 30. August 2013 19:11
>> To: user@synapse.apache.org
>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>
>>
>> On Aug 30, 2013, at 5:40 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>
>>> Hi Hiranja,
>>>
>>> I've done a series of tests that appear to indicate the problem
>>> is in the TCP transport.
>>
>> It's not really a problem in the tcp transport. The issue is exactly what I mentioned in my previous mail. That is, Synapse tries to read the input stream all the way to the end-of-stream, which causes it to get blocked. In addition to your custom mediator, other mediators like <lov level="full"/> also cause the same behavior. Basically any mediator that accesses the full message payload can cause this problem.
>>
>> The way I see it, you have only three options at this point:
>>
>> 1. Change your class mediator to not read the message payload to the end-of-stream. But you still have to get rid of any other mediators that may cause the message to be fully serialized (e.g. <log level="full"/>). Therefore this may not be very practical.
>> 2. Write a custom message builder instead of a class mediator. This is probably the most architecturally sound approach. However, you must make sure that the builder doesn't read the input stream all the way to the end-of-stream. That is, it should read the message up to a known delimiter, or read a known number of bytes. Otherwise you will have the same problem again.
>> 3. Use HTTP to receive the messages (HTTP protocol handles this type of issues using the Content-Length header or the chunked encoding system).
>>
>>> Here are the steps I did and the results I saw:
>>> Important: "still the same" means:
>>> - while the tcp connection is open, there is no reaction from synapse (no log messages etc)
>>> - only AFTER the connection to the TCP listener has been closed do log messages appear and the syslog messages are processed
>>>
>>> 1. Rewrote my class mediator to use BufferedReader.readLine() -> still the same
>>> 2. Rewrote my class mediator to simply replace the payload with fixed messages WITHOUT even reading the existing payload -> still the same
>>> 3. Commented out my class mediator in synapse.xml -> still the same
>>
>> All the above 3 are mainly due to the other serializing mediators in the configuration (e.g. log). Also FYI, changing to the BufferedRead.readLine() doesn't make any difference. That is also a blocking I/O call.
>>
>>> 4. Added transport.tcp.port to the axis2.xml (as well as being in proxyTcp2Jms in synapse.xml) -> error at synapse startup "Error while starting the TCP endpoint. Address already in use"
>>> 5. As above, but removed the port from proxyTcp2Jms in synapse.xml -> error at synapse startup "Service doesn't have configuration information for transport tcp"
>>> 6. Defined transport.tcp.port in both files, but with different port numbers (6060 in synapse.xml, 6061 in axis2.xml) -> synapse starts up OK (with two logfile entries):
>>> - TCPServer.java:76,TCP server started on port : 6061
>>> - TCPServer.java:76,TCP server started on port : 6060
>>> -> when I send a syslog message to port 6060 -> still the same
>>> -> when I send a syslog message to port 6061 -> error written to logfile immediately: AxisEngine.java:219,The service cannot be found for the endpoint reference (EPR)
>>> It appears that the TCPserver configured in synapse.xml is blocking until the client closes the connection,
>>> and the TCPListener configured in axis2.xml is not blocking, but cannot find the proxy definition.
>>
>> That's not how it works. Messages sent to port 6061 must be dispatched by looking at the first element of the SOAP body. In your case it's just a generic <text/> element. Therefore Synapse fails to dispatch the message to any proxy service. This happens before any of the mediators are invoked. Therefore Synapse doesn't get to access the full message payload in this case. Hence it fails early in the pipeline, before Synapse gets blocked on I/O.
>>
>> Thanks,
>> Hiranya
>>
>>>
>>> The tcp jar file I'm using is this one:
>>> http://maven.wso2.org/nexus/content/groups/wso2-public/org/apache/axis2/axis2-transport-tcp/1.1.0-SNAPSHOT/axis2-transport-tcp-1.1.0-SNAPSHOT.jar
>>> but today also tried this one but saw no difference:
>>> http://dist.wso2.org/maven2//org/apache/axis2/axis2-transport-tcp/1.1.0-wso2v5/axis2-transport-tcp-1.1.0-wso2v5.jar
>>>
>>> Thanks,
>>> Peter
>>>
>>> -----Original Message-----
>>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>>> Sent: Montag, 26. August 2013 19:44
>>> To: user@synapse.apache.org
>>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>>
>>> Hi,
>>>
>>> On Aug 26, 2013, at 4:48 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>>
>>>> Hi Hiranja,
>>>>
>>>> No problem. Thanks for the support.
>>>>
>>>> Peter
>>>>
>>>> ===============================================================
>>>>
>>>> /*
>>>> * Filename: syslogMsgBuilder.java
>>>> * Author: P.Wright
>>>> * Date: 06.08.2013
>>>> * Description: Java class for Synapse class mediator.
>>>> * Class is embedded in Synapse, and called when a SYSLOG message is received.
>>>> *
>>>> * The SYSLOG message can contain 1-n syslog messages, each
>>>> * separated by a NL, and the length of each message defined
>>>> * at the start of it. This is called TCP-Framing "octet-counted".
>>>> * See: http://www.rsyslog.com/doc/omfwd.html
>>>> *
>>>> * Example input syslog message containing 3 messages
>>>> * 34 <1> This is dummy message number 1
>>>> * 34 <2> This is dummy message number 2
>>>> * 34 <3> This is dummy message number 3
>>>> *
>>>> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
>>>> * <?xml version='1.0' encoding='utf-8'?>
>>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>>> * <soapenv:Body>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload">
>>>> * 34 <1> This is dummy message number 1
>>>> * 34 <2> This is dummy message number 2
>>>> * 34 <3> This is dummy message number 3
>>>> * </text>
>>>> * </soapenv:Body>
>>>> * </soapenv:Envelope>
>>>> *
>>>> * Actions as follows:
>>>> * - Get the payload
>>>> * - Split the payload on NL
>>>> * - Remove the msglen at the start of each message
>>>> * - Replace any "<" with "<"
>>>> * - Replace the payload with an XML content with 3 messages (see below)
>>>> *
>>>> * Output message as follows:
>>>> * <?xml version='1.0' encoding='utf-8'?>
>>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>>> * <soapenv:Body>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
>>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
>>>> * </soapenv:Body>
>>>> * </soapenv:Envelope>
>>>> *
>>>> * Method Overview:
>>>> * - mediate ("main" method)
>>>> */
>>>> package com.sixtelekurs.classMediators;
>>>>
>>>> import java.util.Iterator;
>>>> import org.apache.axiom.om.OMElement;
>>>> import org.apache.axiom.om.util.AXIOMUtil;
>>>> import org.apache.axiom.soap.SOAPBody;
>>>> import org.apache.axiom.soap.SOAPEnvelope;
>>>> import org.apache.commons.logging.Log;
>>>> import org.apache.commons.logging.LogFactory;
>>>> import org.apache.synapse.MessageContext;
>>>> import org.apache.synapse.Mediator;
>>>>
>>>>
>>>> public class syslogMsgBuilder implements Mediator
>>>> {
>>>> private static String className = "syslogMsgBuilder";
>>>> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
>>>> private String _desc = null;
>>>> private String _uid = null;
>>>>
>>>> // ------------------------------------------------------------------------
>>>> public boolean mediate(MessageContext mc)
>>>> {
>>>> String logMsg = "===" + className + ".mediate called=============";
>>>> log.debug(logMsg);
>>>>
>>>> // Get the message content and process it
>>>> String syslogStr = mc.getEnvelope().getBody().getFirstElement().getText();
>>>
>>> I believe this line if the issue. The call to getText() on the OMSourcedElement probably causes it to read the entire input stream until end-of-stream is encountered. But end-of-stream does not occur until the sender closes its output stream at the remote end.
>>>
>>> I think a proper solution should get the InputStream from the OMSourcedElement and try to read from it until a known number of bytes have been read or a particular delimiter character is encountered. In a case like HTTP, the HTTP message usually have the content-length header or a special delimiter sequence (chunk delimiter) which allows the server to know how much data to read from the input stream without reading all the way to the end-of-stream. In case of TCP transport, there are no special headers to look at, so you must rely on the information available on the message payload it self.
>>>
>>> That's just my take on it. May be there's a better/easier way to solve this problem at Axiom level (perhaps a special method in the OMSourcedElement API?). May be Andreas can shed some light on the matter.
>>>
>>> Thanks,
>>> Hiranya
>>>
>>>> logMsg = "syslogStr='" + syslogStr + "'";
>>>> log.debug(logMsg);
>>>>
>>>> // Delete the existing payload
>>>> logMsg = "Deleting existing payload";
>>>> log.debug(logMsg);
>>>> SOAPBody soapBody = mc.getEnvelope().getBody();
>>>> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
>>>> {
>>>> itr.next();
>>>> itr.remove();
>>>> }
>>>>
>>>> // Split the strings into N syslog messages, convert each to
>>>> // an OMElement, and append to the XML-Body
>>>> int msgCnt = 0;
>>>> try
>>>> {
>>>> // Split the string on NL
>>>> String[] lines = syslogStr.split("\\n");
>>>> logMsg = "syslog.input.cnt(NL)=" + lines.length;
>>>> log.debug(logMsg);
>>>> for (int ii=0; ii < lines.length; ii++)
>>>> {
>>>> // Now split on space (once only) to ignore the msglen
>>>> String[] msg = lines[ii].split(" ", 2);
>>>> if (msg.length == 2)
>>>> {
>>>> // Replace all "<" with "<" for XML
>>>> String syslogMsg = msg[1].replaceAll("<", "<");
>>>> int jj = ii+1;
>>>> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
>>>> log.debug(logMsg);
>>>> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
>>>> xmlStr += syslogMsg;
>>>> xmlStr += "</text>";
>>>> msgCnt++;
>>>>
>>>> // Now add elem
>>>> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
>>>> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
>>>> log.debug(logMsg);
>>>> soapBody.addChild(xmlElem);
>>>> }
>>>> else
>>>> {
>>>> logMsg = "msg could not be split correctly [" + msg + "]";
>>>> log.error(logMsg);
>>>> }
>>>> }
>>>> }
>>>> catch (Exception e)
>>>> {
>>>> logMsg = "ERROR: " + e;
>>>> log.error(logMsg);
>>>> return(false);
>>>> }
>>>>
>>>> if (msgCnt == 0)
>>>> {
>>>> logMsg = "ERROR: no messages processed";
>>>> log.error(logMsg);
>>>> return(false);
>>>> }
>>>>
>>>> logMsg = "Made " + msgCnt + " messages";
>>>> log.info(logMsg);
>>>>
>>>> SOAPEnvelope envelope = mc.getEnvelope();
>>>> logMsg = "SOAP.Envelope: " + envelope;
>>>> log.debug(logMsg);
>>>>
>>>> return(true);
>>>> }
>>>>
>>>> // SET methods --------------------------------
>>>> public String getDescription()
>>>> {
>>>> String msg = className + ".desc";
>>>> return(msg);
>>>> }
>>>> public String getType()
>>>> {
>>>> String msg = className + ".type";
>>>> return(msg);
>>>> }
>>>> public int getTraceState()
>>>> {
>>>> return(0);
>>>> }
>>>>
>>>> // SET methods --------------------------------
>>>> public void setTraceState(int traceState)
>>>> {
>>>> traceState = 0;
>>>> }
>>>> public void setDescription(String desc)
>>>> {
>>>> _desc = desc;
>>>> }
>>>> public void setUid(String uid)
>>>> {
>>>> _uid = uid;
>>>> }
>>>> }
>>>>
>>>> -----Original Message-----
>>>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>>>> Sent: Freitag, 23. August 2013 08:30
>>>> To: user@synapse.apache.org
>>>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>>>
>>>> It sounds like something is trying to read from the input stream until the end of stream is encountered. This may be the plain text builder or your custom mediator. Can you share your custom mediator source code so we can take a look?
>>>>
>>>> Thanks,
>>>> Hiranya
>>>>
>>>> On Aug 22, 2013, at 8:27 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>>>
>>>>> Hi,
>>>>>
>>>>> I have my TCP Syslog --> JMS proxy up and running (see previous emails), but now have encountered another problem.
>>>>> Up till now I have simply tested by using netstat to send 1-n messages stored in a file to the TCP server.
>>>>> In this case, the messages are being received and processed correctly by synapse (sent as JMS
>>>>> messages to the JMS server). It seems that this is working correctly, because the connection to the
>>>>> synapse TCP server is being closed by netstat after it sends the messages.
>>>>>
>>>>> In my real world case however, we have 1-n clients writing syslog messages to a
>>>>> local syslog server. This syslog server has been configured to send the messages on
>>>>> to the (remote) synapse TCP server. In this case the messages are NOT being received
>>>>> and processed by synapse. Only after the syslog server closes it connection to the synapse
>>>>> TCP server, are the messages being received and processed by synapse.
>>>>>
>>>>> Any ideas what the problem could be?
>>>>> Attached again my synapse.xml config file.
>>>>>
>>>>> Thanks,
>>>>> Peter
>>>>>
>>>>> ===============================================
>>>>>
>>>>> <definitions xmlns="http://ws.apache.org/ns/synapse">
>>>>> <sequence name="fault">
>>>>> <makefault>
>>>>> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
>>>>> <reason value="Mediation failed."/>
>>>>> </makefault>
>>>>> <send/>
>>>>> </sequence>
>>>>> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
>>>>> <in>
>>>>> <log level="full"/>
>>>>> <send/>
>>>>> </in>
>>>>> <out>
>>>>> <send/>
>>>>> </out>
>>>>> </sequence>
>>>>>
>>>>> <proxy name="proxyTcp2Jms" transports="tcp">
>>>>> <target>
>>>>> <inSequence>
>>>>> <!-- Define TCP listener expects plain text (syslog) messages -->
>>>>> <property name="messageType" value="text/plain" scope="axis2"/>
>>>>> <property name="OUT_ONLY" value="true"/>
>>>>> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
>>>>> <log level="full"/>
>>>>>
>>>>> <!-- Split any multiple syslog messages for the iterator -->
>>>>> <!-- and return as XML with child messages -->
>>>>> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
>>>>> <log level="full"/>
>>>>> </class>
>>>>>
>>>>> <log level="full"/>
>>>>>
>>>>> <!-- Iterate over any multiple messages -->
>>>>> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
>>>>> <target>
>>>>> <sequence>
>>>>> <send>
>>>>> <endpoint>
>>>>> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
>>>>> </endpoint>
>>>>> </send>
>>>>> </sequence>
>>>>> </target>
>>>>> </iterate>
>>>>> </inSequence>
>>>>> <outSequence/>
>>>>> <faultSequence>
>>>>> <log level="full" category="ERROR" separator=","/>
>>>>> </faultSequence>
>>>>> </target>
>>>>> <parameter name="transport.tcp.port">6060</parameter>
>>>>> <parameter name="transport.tcp.contentType">text/plain</parameter>
>>>>> </proxy>
>>>>>
>>>>> </definitions>
>>>>>
>>>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>>>> Thank you.
>>>>
>>>> --
>>>> Hiranya Jayathilaka
>>>> Mayhem Lab/RACE Lab;
>>>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>>>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>>>> Blog: http://techfeast-hiranya.blogspot.com
>>>>
>>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>>> Thank you.
>>>
>>> --
>>> Hiranya Jayathilaka
>>> Mayhem Lab/RACE Lab;
>>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>>> Blog: http://techfeast-hiranya.blogspot.com
>>>
>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>> Thank you.
>>
>> --
>> Hiranya Jayathilaka
>> Mayhem Lab/RACE Lab;
>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>> Blog: http://techfeast-hiranya.blogspot.com
>>
>> The content of this e-mail is intended only for the confidential use of the person addressed.
>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>> Thank you.
>
> --
> Hiranya Jayathilaka
> Mayhem Lab/RACE Lab;
> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
> Blog: http://techfeast-hiranya.blogspot.com
>
> The content of this e-mail is intended only for the confidential use of the person addressed.
> If you are not the intended recipient, please notify the sender and delete this email immediately.
> Thank you.
--
Hiranya Jayathilaka
Mayhem Lab/RACE Lab;
Dept. of Computer Science, UCSB; http://cs.ucsb.edu
E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
Blog: http://techfeast-hiranya.blogspot.com
RE: Synapse: TCP syslog to JMS proxy: messages only processed in
synapse after client TCP connection is closed
Posted by "Wright, Peter" <Pe...@six-group.com>.
Hi Hiranja,
I'm still having a problem with finding "end of synapse message".
A synapse message can contain 1-n syslog messages.
Each syslog message is preceeded by its message length,
so I have no problem reading each message.
The problem is to know when to stop trying to read a possible
next message i.e. end of synapse message (end of frame?).
When I try to read one byte of the length of the next
(non existing) message, the read blocks.
Every message has a terminating newline character, which
is not necessary, as the message length is defined anyway.
If the newline were only at the end of frame (synapse message),
I could use that to know I should stop reading.
Who is adding the newlines? The syslog sender, or the TCP transport
module of synapse? Any chance of finding a solution here?
(end of frame delimiter?).
> TCP transport will inject all messages sent on the same connection as a single message into Synapse.
Does the TCP transport buffer the input syslog messages before concatenating them into the synapse message?
When are the messages sent? Buffer size reached? Time? Hopefully not only after the TCP connection is closed??
Is one synapse message equivalent to one TCP frame?
In our test setup, we have a syslog server configured to send all received messages
on to the synapse TCP listener.
Thanks
Peter
-----Original Message-----
From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
Sent: Donnerstag, 5. September 2013 19:59
To: user@synapse.apache.org
Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
You need to somehow figure out the number of messages to read from the input stream and read all those messages in your custom mediator. TCP transport will inject all messages sent on the same connection as a single message into Synapse.
If you want to debug your code, start synapse with the flag -xdebug and do a remote debug via port 8000.
Thanks,
Hiranya
On Sep 5, 2013, at 5:34 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
> Hi Hiranja,
>
> I've rewritten my class mediator to just read the message content according to the message lengths,
> and changed the log mediator settings as you suggested, but there still appears to be a problem.
>
> When a client writes 3 separate messages, opening and closing the TCP connection each time, all messages
> are received and processed correctly by synapse.
> However, when a client opens a TCP connection, writes 3 separate messages, and then closes the connection,
> only the first message is received and processed correctly by synapse.
>
> My class mediator processes the first message correctly, so it appears(?) to be processing and returning
> the message content correctly (without blocking), but there are no log entries at all for the following
> messages.
>
> I've attached my various config files and the source code of my class mediator.
>
> Thanks for your help,
> Peter
>
> ======================================
>
>
> synapse.log
> -----------
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,MediatorLog.java:97,To: , WSAction: urn:mediate, SOAPAction: urn:mediate, MessageID: urn:uuid:B3133E5C6DF9C641551378383293591, Direction: request
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:78,===syslogMsgBuilder.mediate called=============
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:161,getSyslogMsgs. start
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:171,getSyslogMsgs. Reading msg 1 =============
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:229,readMsgLen. start
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.0=1
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.1=9
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.2=4
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:260,readMsgLen: strMsglen = '194'
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:283,readMsgLen: intMsglen = 194
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:299,readMsgContent. start. readLen=194
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:342,readMsgContent: msgContent=[<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] ], nlFound=false
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:195,getSyslogMsgs: no NL found: stop
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:85,Found 1 message(s). Delete existing payload
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:105,syslogMsg.1='<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] '
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:113,xmlStr.1='<text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text>'
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:132,Made 1 message(s)
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:136,SOAP.Envelope: <?xml version='1.0' encoding='utf-8'?><soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"><soapenv:Body><text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text></soapenv:Body></soapenv:Envelope>
>
> synapse.xml
> -----------
> <definitions xmlns="http://ws.apache.org/ns/synapse">
> <sequence name="fault">
> <makefault>
> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
> <reason value="Mediation failed."/>
> </makefault>
> <send/>
> </sequence>
> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
> <in>
> <log level="simple"/>
> <send/>
> </in>
> <out>
> <send/>
> </out>
> </sequence>
>
> <proxy name="proxyTcp2Jms" transports="tcp">
> <target>
> <inSequence>
> <!-- Define TCP listener expects plain text (syslog) messages -->
> <property name="messageType" value="text/plain" scope="axis2"/>
> <property name="OUT_ONLY" value="true"/>
> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
> <log level="simple"/> <!-- simple,headers,full -->
>
> <!-- Split any multiple syslog messages for the iterator -->
> <!-- and return as XML with child messages -->
> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
> <!-- <log level="full"/> -->
> </class>
>
> <!-- <log level="full"/> -->
>
> <!-- Iterate over any multiple messages -->
> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
> <target>
> <sequence>
> <send>
> <!-- <log level="full"/> -->
> <endpoint>
> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
> </endpoint>
> </send>
> </sequence>
> </target>
> </iterate>
> </inSequence>
> <!-- OUT_ONLY=true
> <outSequence>
> <log level="simple" category="INFO" separator=","/>
> <send/>
> </outSequence>
> -->
> <faultSequence>
> <log level="simple" category="ERROR" separator=","/>
> </faultSequence>
> </target>
> <parameter name="transport.tcp.port">6060</parameter>
> <parameter name="transport.tcp.contentType">text/plain</parameter>
> <!-- <parameter name="transport.tcp.backlog">0</parameter> -->
> </proxy>
> </definitions>
>
>
> syslogMsgBuilder.java
> ---------------------
> /*
> * Filename: syslogMsgBuilder.java
> * Author: P.Wright
> * Date: 06.08.2013
> * Description: Java class for Synapse class mediator.
> * Class is embedded in Synapse, and called when a SYSLOG message is received.
> *
> * The SYSLOG message can contain 1-n syslog messages, each
> * separated by a NL, and the length of each message defined
> * at the start of it. This is called TCP-Framing "octet-counted".
> * See: http://www.rsyslog.com/doc/omfwd.html
> *
> * Example input syslog message containing 3 messages
> * 35 <1> This is dummy message number 1\n
> * 35 <2> This is dummy message number 2\n
> * 34 <3> This is dummy message number 3
> *
> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
> * <?xml version='1.0' encoding='utf-8'?>
> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
> * <soapenv:Body>
> * <text xmlns="http://ws.apache.org/commons/ns/payload">
> * 35 <1> This is dummy message number 1\n
> * 35 <2> This is dummy message number 2\n
> * 34 <3> This is dummy message number 3
> * </text>
> * </soapenv:Body>
> * </soapenv:Envelope>
> *
> * Actions as follows:
> * - Get the first element of the message body
> * - Get a reader to the content of the first element
> * - Read the msglen
> * - Read the message according to msglen
> * - If last character is NOT a '\n', stop reading messages
> * - If last character IS a '\n', keep going and read next msglen
> * - Replace any "<" with "<"
> * - Replace the payload with an XML content with 3 messages (see below)
> *
> * Output message as follows:
> * <?xml version='1.0' encoding='utf-8'?>
> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
> * <soapenv:Body>
> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
> * </soapenv:Body>
> * </soapenv:Envelope>
> *
> * Method Overview:
> * - mediate ("main" method)
> */
>
> package com.sixtelekurs.classMediators;
>
> import java.io.Reader;
> import java.util.Iterator;
> import java.util.ArrayList;
> import java.util.List;
> import org.apache.axiom.om.OMElement;
> import org.apache.axiom.om.util.AXIOMUtil;
> import org.apache.axiom.soap.SOAPBody;
> import org.apache.axiom.soap.SOAPEnvelope;
> import org.apache.commons.logging.Log;
> import org.apache.commons.logging.LogFactory;
> import org.apache.synapse.MessageContext;
> import org.apache.synapse.Mediator;
>
>
> public class syslogMsgBuilder implements Mediator
> {
> private static String className = "syslogMsgBuilder";
> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
> private String _desc = null;
> private String _uid = null;
>
> // ------------------------------------------------------------------------
> public boolean mediate(MessageContext mc)
> {
> String logMsg = "===" + className + ".mediate called=============";
> log.info(logMsg);
>
> // Get the message content as an array of strings
> List<String> listSyslogStr = getSyslogMsgs(mc.getEnvelope().getBody().getFirstElement());
>
> // Delete the existing payload
> logMsg = "Found " + listSyslogStr.size() + " message(s). Delete existing payload";
> log.info(logMsg);
> SOAPBody soapBody = mc.getEnvelope().getBody();
> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
> {
> itr.next();
> itr.remove();
> }
>
> // Traverse the string list, convert each to an XML element
> // and write into the SOAP body
> int msgCnt = 0;
> try
> {
> // Get the syslog msg
> for (String msg : listSyslogStr)
> {
> // Replace all "<" with "<" for XML
> String syslogMsg = msg.replaceAll("<", "<");
> int jj = msgCnt+1;
> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
> log.debug(logMsg);
> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
> xmlStr += syslogMsg;
> xmlStr += "</text>";
> msgCnt++;
>
> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
> log.info(logMsg);
> soapBody.addChild(xmlElem);
> }
> }
> catch (Exception e)
> {
> logMsg = "ERROR: " + e;
> log.error(logMsg);
> return(false);
> }
>
> if (msgCnt == 0)
> {
> logMsg = "ERROR: no messages processed";
> log.error(logMsg);
> return(false);
> }
>
> logMsg = "Made " + msgCnt + " message(s)";
> log.info(logMsg);
>
> SOAPEnvelope envelope = mc.getEnvelope();
> logMsg = "SOAP.Envelope: " + envelope;
> log.info(logMsg);
>
> return(true);
> }
>
> class ReturnValues
> {
> String msgString;
> boolean foundNewline;
> ReturnValues(final String msgString, boolean foundNewline)
> {
> this.msgString = msgString;
> this.foundNewline = foundNewline;
> }
> }
>
> // getSyslogMsgs: traverses the input message
> // and returns any syslog messages as an array
> // of strings in XML form
> private List<String> getSyslogMsgs(OMElement omElem)
> {
> String funcname = "getSyslogMsgs";
> String logMsg = "";
>
> logMsg = funcname + ". start";
> log.debug(logMsg);
>
> List<String> syslogMsgs = new ArrayList<String>();
> Reader reader = omElem.getTextAsStream(false); // pass cache=false
>
> int msgCnt = 0;
> boolean keepGoing = true;
> while (keepGoing)
> {
> logMsg = funcname + ". Reading msg " + (msgCnt+1) + " =============";
> log.info(logMsg);
>
> // Get the msglen
> int msgLen = readMsgLen(reader);
> if (msgLen <= 0)
> {
> keepGoing = false;
> }
> else
> {
> // Get the message
> ReturnValues retvals = readMsgContent(reader, msgLen);
> if (retvals.msgString.isEmpty())
> {
> keepGoing = false;
> }
> else
> {
> syslogMsgs.add(retvals.msgString);
>
> // Read any trailing NL
> if (!retvals.foundNewline)
> {
> logMsg = funcname + ": no NL found: stop";
> log.debug(logMsg);
> keepGoing = false;
> }
> }
> }
> msgCnt++;
> }
>
> // Close the reader
> /*********************
> try
> {
> reader.close();
> }
> catch (Exception e)
> {
> logMsg = funcname + ": reader.close failed. " + e;
> log.error(logMsg);
> }
> *********************/
> return(syslogMsgs);
> }
>
> // Get the msglen of the syslog message
> // Read characters till a space and convert to an integer
> // Return the int msglen
> private int readMsgLen(Reader reader)
> {
> String funcname = "readMsgLen";
> int intMsglen = 0;
> String logMsg = "";
> String strMsglen = "";
>
> logMsg = funcname + ". start";
> log.debug(logMsg);
>
> try
> {
> char[] cBuffer = new char[10];
> int cCnt=0;
> boolean keepGoing = true;
> while (keepGoing)
> {
> // Read one character at a time into the char array
> int rc = reader.read(cBuffer, cCnt, 1);
> if (rc <= 0)
> {
> keepGoing = false;
> }
> else if (cBuffer[cCnt] != ' ')
> {
> logMsg = "cBuffer." + cCnt + "=" + cBuffer[cCnt];
> log.debug(logMsg);
> cCnt++;
> }
> else
> {
> keepGoing = false;
> }
> }
> if (cCnt > 0)
> {
> // Convert the characters read into a string
> strMsglen = new String(cBuffer, 0, cCnt);
> logMsg = funcname + ": strMsglen = '" + strMsglen + "'";
> log.debug(logMsg);
> }
> }
> catch (Exception e)
> {
> logMsg = funcname + ": error reading msglen" + e;
> log.error(logMsg);
> }
>
> if (strMsglen != "")
> {
> // Convert the (string)msglen to an (int)msglen
> try
> {
> intMsglen = Integer.parseInt(strMsglen);
> }
> catch (Exception e)
> {
> logMsg = funcname + ": error converting string '" + strMsglen + "' to integer. " + e;
> log.error(logMsg);
> }
> }
> logMsg = funcname + ": intMsglen = " + intMsglen;
> log.debug(logMsg);
> return(intMsglen);
> }
>
> // readMsgContent --------------------------------------------
> private ReturnValues readMsgContent(Reader reader, int msgLen)
> {
> String funcname = "readMsgContent";
> String msgContent = "";
> boolean nlFound = false;
> String logMsg = "";
> int maxlen = 2048;
> char[] cBuffer = new char[maxlen];
> int readLen = (msgLen < (maxlen-1) ? msgLen:(maxlen-1));
>
> logMsg = funcname + ". start. readLen=" + readLen;
> log.debug(logMsg);
>
> if (msgLen > maxlen-1)
> {
> logMsg = "message too long for buffer (" + msgLen + "/" + (maxlen-1);
> log.error(logMsg);
> ReturnValues retvals = new ReturnValues("", false);
> return(retvals);
> }
>
> try
> {
> int rc = reader.read(cBuffer, 0, readLen);
> if (rc <= 0)
> {
> logMsg = funcname + ": error reading msg: " + rc;
> log.error(logMsg);
> }
> else
> {
> if (rc < readLen)
> {
> logMsg = "read: IST=" + rc + ", SOLL=" + readLen;
> log.warn(logMsg);
> }
>
> // Ignore any trailing NL
> if (cBuffer[rc-1] == '\n')
> {
> nlFound = true;
> logMsg = "NL read at end of message - ignore it";
> log.debug(logMsg);
> rc--;
> }
> msgContent = new String(cBuffer, 0, rc);
> }
> }
> catch (Exception e)
> {
> logMsg = funcname + ": exception reading msg" + e;
> log.error(logMsg);
> }
> logMsg = funcname + ": msgContent=[" + msgContent + "], nlFound=" + nlFound;
> log.debug(logMsg);
>
> ReturnValues retvals = new ReturnValues(msgContent, nlFound);
> return(retvals);
> }
>
> // SET methods --------------------------------
> public String getDescription()
> {
> String msg = className + ".desc";
> return(msg);
> }
> public String getType()
> {
> String msg = className + ".type";
> return(msg);
> }
> public int getTraceState()
> {
> return(0);
> }
>
> // SET methods --------------------------------
> public void setTraceState(int traceState)
> {
> traceState = 0;
> }
> public void setDescription(String desc)
> {
> _desc = desc;
> }
> public void setUid(String uid)
> {
> _uid = uid;
> }
> }
>
> -----Original Message-----
> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
> Sent: Freitag, 30. August 2013 19:11
> To: user@synapse.apache.org
> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>
>
> On Aug 30, 2013, at 5:40 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>
>> Hi Hiranja,
>>
>> I've done a series of tests that appear to indicate the problem
>> is in the TCP transport.
>
> It's not really a problem in the tcp transport. The issue is exactly what I mentioned in my previous mail. That is, Synapse tries to read the input stream all the way to the end-of-stream, which causes it to get blocked. In addition to your custom mediator, other mediators like <lov level="full"/> also cause the same behavior. Basically any mediator that accesses the full message payload can cause this problem.
>
> The way I see it, you have only three options at this point:
>
> 1. Change your class mediator to not read the message payload to the end-of-stream. But you still have to get rid of any other mediators that may cause the message to be fully serialized (e.g. <log level="full"/>). Therefore this may not be very practical.
> 2. Write a custom message builder instead of a class mediator. This is probably the most architecturally sound approach. However, you must make sure that the builder doesn't read the input stream all the way to the end-of-stream. That is, it should read the message up to a known delimiter, or read a known number of bytes. Otherwise you will have the same problem again.
> 3. Use HTTP to receive the messages (HTTP protocol handles this type of issues using the Content-Length header or the chunked encoding system).
>
>> Here are the steps I did and the results I saw:
>> Important: "still the same" means:
>> - while the tcp connection is open, there is no reaction from synapse (no log messages etc)
>> - only AFTER the connection to the TCP listener has been closed do log messages appear and the syslog messages are processed
>>
>> 1. Rewrote my class mediator to use BufferedReader.readLine() -> still the same
>> 2. Rewrote my class mediator to simply replace the payload with fixed messages WITHOUT even reading the existing payload -> still the same
>> 3. Commented out my class mediator in synapse.xml -> still the same
>
> All the above 3 are mainly due to the other serializing mediators in the configuration (e.g. log). Also FYI, changing to the BufferedRead.readLine() doesn't make any difference. That is also a blocking I/O call.
>
>> 4. Added transport.tcp.port to the axis2.xml (as well as being in proxyTcp2Jms in synapse.xml) -> error at synapse startup "Error while starting the TCP endpoint. Address already in use"
>> 5. As above, but removed the port from proxyTcp2Jms in synapse.xml -> error at synapse startup "Service doesn't have configuration information for transport tcp"
>> 6. Defined transport.tcp.port in both files, but with different port numbers (6060 in synapse.xml, 6061 in axis2.xml) -> synapse starts up OK (with two logfile entries):
>> - TCPServer.java:76,TCP server started on port : 6061
>> - TCPServer.java:76,TCP server started on port : 6060
>> -> when I send a syslog message to port 6060 -> still the same
>> -> when I send a syslog message to port 6061 -> error written to logfile immediately: AxisEngine.java:219,The service cannot be found for the endpoint reference (EPR)
>> It appears that the TCPserver configured in synapse.xml is blocking until the client closes the connection,
>> and the TCPListener configured in axis2.xml is not blocking, but cannot find the proxy definition.
>
> That's not how it works. Messages sent to port 6061 must be dispatched by looking at the first element of the SOAP body. In your case it's just a generic <text/> element. Therefore Synapse fails to dispatch the message to any proxy service. This happens before any of the mediators are invoked. Therefore Synapse doesn't get to access the full message payload in this case. Hence it fails early in the pipeline, before Synapse gets blocked on I/O.
>
> Thanks,
> Hiranya
>
>>
>> The tcp jar file I'm using is this one:
>> http://maven.wso2.org/nexus/content/groups/wso2-public/org/apache/axis2/axis2-transport-tcp/1.1.0-SNAPSHOT/axis2-transport-tcp-1.1.0-SNAPSHOT.jar
>> but today also tried this one but saw no difference:
>> http://dist.wso2.org/maven2//org/apache/axis2/axis2-transport-tcp/1.1.0-wso2v5/axis2-transport-tcp-1.1.0-wso2v5.jar
>>
>> Thanks,
>> Peter
>>
>> -----Original Message-----
>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>> Sent: Montag, 26. August 2013 19:44
>> To: user@synapse.apache.org
>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>
>> Hi,
>>
>> On Aug 26, 2013, at 4:48 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>
>>> Hi Hiranja,
>>>
>>> No problem. Thanks for the support.
>>>
>>> Peter
>>>
>>> ===============================================================
>>>
>>> /*
>>> * Filename: syslogMsgBuilder.java
>>> * Author: P.Wright
>>> * Date: 06.08.2013
>>> * Description: Java class for Synapse class mediator.
>>> * Class is embedded in Synapse, and called when a SYSLOG message is received.
>>> *
>>> * The SYSLOG message can contain 1-n syslog messages, each
>>> * separated by a NL, and the length of each message defined
>>> * at the start of it. This is called TCP-Framing "octet-counted".
>>> * See: http://www.rsyslog.com/doc/omfwd.html
>>> *
>>> * Example input syslog message containing 3 messages
>>> * 34 <1> This is dummy message number 1
>>> * 34 <2> This is dummy message number 2
>>> * 34 <3> This is dummy message number 3
>>> *
>>> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
>>> * <?xml version='1.0' encoding='utf-8'?>
>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>> * <soapenv:Body>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload">
>>> * 34 <1> This is dummy message number 1
>>> * 34 <2> This is dummy message number 2
>>> * 34 <3> This is dummy message number 3
>>> * </text>
>>> * </soapenv:Body>
>>> * </soapenv:Envelope>
>>> *
>>> * Actions as follows:
>>> * - Get the payload
>>> * - Split the payload on NL
>>> * - Remove the msglen at the start of each message
>>> * - Replace any "<" with "<"
>>> * - Replace the payload with an XML content with 3 messages (see below)
>>> *
>>> * Output message as follows:
>>> * <?xml version='1.0' encoding='utf-8'?>
>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>> * <soapenv:Body>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
>>> * </soapenv:Body>
>>> * </soapenv:Envelope>
>>> *
>>> * Method Overview:
>>> * - mediate ("main" method)
>>> */
>>> package com.sixtelekurs.classMediators;
>>>
>>> import java.util.Iterator;
>>> import org.apache.axiom.om.OMElement;
>>> import org.apache.axiom.om.util.AXIOMUtil;
>>> import org.apache.axiom.soap.SOAPBody;
>>> import org.apache.axiom.soap.SOAPEnvelope;
>>> import org.apache.commons.logging.Log;
>>> import org.apache.commons.logging.LogFactory;
>>> import org.apache.synapse.MessageContext;
>>> import org.apache.synapse.Mediator;
>>>
>>>
>>> public class syslogMsgBuilder implements Mediator
>>> {
>>> private static String className = "syslogMsgBuilder";
>>> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
>>> private String _desc = null;
>>> private String _uid = null;
>>>
>>> // ------------------------------------------------------------------------
>>> public boolean mediate(MessageContext mc)
>>> {
>>> String logMsg = "===" + className + ".mediate called=============";
>>> log.debug(logMsg);
>>>
>>> // Get the message content and process it
>>> String syslogStr = mc.getEnvelope().getBody().getFirstElement().getText();
>>
>> I believe this line if the issue. The call to getText() on the OMSourcedElement probably causes it to read the entire input stream until end-of-stream is encountered. But end-of-stream does not occur until the sender closes its output stream at the remote end.
>>
>> I think a proper solution should get the InputStream from the OMSourcedElement and try to read from it until a known number of bytes have been read or a particular delimiter character is encountered. In a case like HTTP, the HTTP message usually have the content-length header or a special delimiter sequence (chunk delimiter) which allows the server to know how much data to read from the input stream without reading all the way to the end-of-stream. In case of TCP transport, there are no special headers to look at, so you must rely on the information available on the message payload it self.
>>
>> That's just my take on it. May be there's a better/easier way to solve this problem at Axiom level (perhaps a special method in the OMSourcedElement API?). May be Andreas can shed some light on the matter.
>>
>> Thanks,
>> Hiranya
>>
>>> logMsg = "syslogStr='" + syslogStr + "'";
>>> log.debug(logMsg);
>>>
>>> // Delete the existing payload
>>> logMsg = "Deleting existing payload";
>>> log.debug(logMsg);
>>> SOAPBody soapBody = mc.getEnvelope().getBody();
>>> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
>>> {
>>> itr.next();
>>> itr.remove();
>>> }
>>>
>>> // Split the strings into N syslog messages, convert each to
>>> // an OMElement, and append to the XML-Body
>>> int msgCnt = 0;
>>> try
>>> {
>>> // Split the string on NL
>>> String[] lines = syslogStr.split("\\n");
>>> logMsg = "syslog.input.cnt(NL)=" + lines.length;
>>> log.debug(logMsg);
>>> for (int ii=0; ii < lines.length; ii++)
>>> {
>>> // Now split on space (once only) to ignore the msglen
>>> String[] msg = lines[ii].split(" ", 2);
>>> if (msg.length == 2)
>>> {
>>> // Replace all "<" with "<" for XML
>>> String syslogMsg = msg[1].replaceAll("<", "<");
>>> int jj = ii+1;
>>> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
>>> log.debug(logMsg);
>>> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
>>> xmlStr += syslogMsg;
>>> xmlStr += "</text>";
>>> msgCnt++;
>>>
>>> // Now add elem
>>> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
>>> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
>>> log.debug(logMsg);
>>> soapBody.addChild(xmlElem);
>>> }
>>> else
>>> {
>>> logMsg = "msg could not be split correctly [" + msg + "]";
>>> log.error(logMsg);
>>> }
>>> }
>>> }
>>> catch (Exception e)
>>> {
>>> logMsg = "ERROR: " + e;
>>> log.error(logMsg);
>>> return(false);
>>> }
>>>
>>> if (msgCnt == 0)
>>> {
>>> logMsg = "ERROR: no messages processed";
>>> log.error(logMsg);
>>> return(false);
>>> }
>>>
>>> logMsg = "Made " + msgCnt + " messages";
>>> log.info(logMsg);
>>>
>>> SOAPEnvelope envelope = mc.getEnvelope();
>>> logMsg = "SOAP.Envelope: " + envelope;
>>> log.debug(logMsg);
>>>
>>> return(true);
>>> }
>>>
>>> // SET methods --------------------------------
>>> public String getDescription()
>>> {
>>> String msg = className + ".desc";
>>> return(msg);
>>> }
>>> public String getType()
>>> {
>>> String msg = className + ".type";
>>> return(msg);
>>> }
>>> public int getTraceState()
>>> {
>>> return(0);
>>> }
>>>
>>> // SET methods --------------------------------
>>> public void setTraceState(int traceState)
>>> {
>>> traceState = 0;
>>> }
>>> public void setDescription(String desc)
>>> {
>>> _desc = desc;
>>> }
>>> public void setUid(String uid)
>>> {
>>> _uid = uid;
>>> }
>>> }
>>>
>>> -----Original Message-----
>>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>>> Sent: Freitag, 23. August 2013 08:30
>>> To: user@synapse.apache.org
>>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>>
>>> It sounds like something is trying to read from the input stream until the end of stream is encountered. This may be the plain text builder or your custom mediator. Can you share your custom mediator source code so we can take a look?
>>>
>>> Thanks,
>>> Hiranya
>>>
>>> On Aug 22, 2013, at 8:27 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>>
>>>> Hi,
>>>>
>>>> I have my TCP Syslog --> JMS proxy up and running (see previous emails), but now have encountered another problem.
>>>> Up till now I have simply tested by using netstat to send 1-n messages stored in a file to the TCP server.
>>>> In this case, the messages are being received and processed correctly by synapse (sent as JMS
>>>> messages to the JMS server). It seems that this is working correctly, because the connection to the
>>>> synapse TCP server is being closed by netstat after it sends the messages.
>>>>
>>>> In my real world case however, we have 1-n clients writing syslog messages to a
>>>> local syslog server. This syslog server has been configured to send the messages on
>>>> to the (remote) synapse TCP server. In this case the messages are NOT being received
>>>> and processed by synapse. Only after the syslog server closes it connection to the synapse
>>>> TCP server, are the messages being received and processed by synapse.
>>>>
>>>> Any ideas what the problem could be?
>>>> Attached again my synapse.xml config file.
>>>>
>>>> Thanks,
>>>> Peter
>>>>
>>>> ===============================================
>>>>
>>>> <definitions xmlns="http://ws.apache.org/ns/synapse">
>>>> <sequence name="fault">
>>>> <makefault>
>>>> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
>>>> <reason value="Mediation failed."/>
>>>> </makefault>
>>>> <send/>
>>>> </sequence>
>>>> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
>>>> <in>
>>>> <log level="full"/>
>>>> <send/>
>>>> </in>
>>>> <out>
>>>> <send/>
>>>> </out>
>>>> </sequence>
>>>>
>>>> <proxy name="proxyTcp2Jms" transports="tcp">
>>>> <target>
>>>> <inSequence>
>>>> <!-- Define TCP listener expects plain text (syslog) messages -->
>>>> <property name="messageType" value="text/plain" scope="axis2"/>
>>>> <property name="OUT_ONLY" value="true"/>
>>>> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
>>>> <log level="full"/>
>>>>
>>>> <!-- Split any multiple syslog messages for the iterator -->
>>>> <!-- and return as XML with child messages -->
>>>> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
>>>> <log level="full"/>
>>>> </class>
>>>>
>>>> <log level="full"/>
>>>>
>>>> <!-- Iterate over any multiple messages -->
>>>> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
>>>> <target>
>>>> <sequence>
>>>> <send>
>>>> <endpoint>
>>>> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
>>>> </endpoint>
>>>> </send>
>>>> </sequence>
>>>> </target>
>>>> </iterate>
>>>> </inSequence>
>>>> <outSequence/>
>>>> <faultSequence>
>>>> <log level="full" category="ERROR" separator=","/>
>>>> </faultSequence>
>>>> </target>
>>>> <parameter name="transport.tcp.port">6060</parameter>
>>>> <parameter name="transport.tcp.contentType">text/plain</parameter>
>>>> </proxy>
>>>>
>>>> </definitions>
>>>>
>>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>>> Thank you.
>>>
>>> --
>>> Hiranya Jayathilaka
>>> Mayhem Lab/RACE Lab;
>>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>>> Blog: http://techfeast-hiranya.blogspot.com
>>>
>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>> Thank you.
>>
>> --
>> Hiranya Jayathilaka
>> Mayhem Lab/RACE Lab;
>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>> Blog: http://techfeast-hiranya.blogspot.com
>>
>> The content of this e-mail is intended only for the confidential use of the person addressed.
>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>> Thank you.
>
> --
> Hiranya Jayathilaka
> Mayhem Lab/RACE Lab;
> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
> Blog: http://techfeast-hiranya.blogspot.com
>
> The content of this e-mail is intended only for the confidential use of the person addressed.
> If you are not the intended recipient, please notify the sender and delete this email immediately.
> Thank you.
--
Hiranya Jayathilaka
Mayhem Lab/RACE Lab;
Dept. of Computer Science, UCSB; http://cs.ucsb.edu
E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
Blog: http://techfeast-hiranya.blogspot.com
The content of this e-mail is intended only for the confidential use of the person addressed.
If you are not the intended recipient, please notify the sender and delete this email immediately.
Thank you.
Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
Posted by Hiranya Jayathilaka <hi...@gmail.com>.
You need to somehow figure out the number of messages to read from the input stream and read all those messages in your custom mediator. TCP transport will inject all messages sent on the same connection as a single message into Synapse.
If you want to debug your code, start synapse with the flag -xdebug and do a remote debug via port 8000.
Thanks,
Hiranya
On Sep 5, 2013, at 5:34 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
> Hi Hiranja,
>
> I've rewritten my class mediator to just read the message content according to the message lengths,
> and changed the log mediator settings as you suggested, but there still appears to be a problem.
>
> When a client writes 3 separate messages, opening and closing the TCP connection each time, all messages
> are received and processed correctly by synapse.
> However, when a client opens a TCP connection, writes 3 separate messages, and then closes the connection,
> only the first message is received and processed correctly by synapse.
>
> My class mediator processes the first message correctly, so it appears(?) to be processing and returning
> the message content correctly (without blocking), but there are no log entries at all for the following
> messages.
>
> I've attached my various config files and the source code of my class mediator.
>
> Thanks for your help,
> Peter
>
> ======================================
>
>
> synapse.log
> -----------
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,MediatorLog.java:97,To: , WSAction: urn:mediate, SOAPAction: urn:mediate, MessageID: urn:uuid:B3133E5C6DF9C641551378383293591, Direction: request
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:78,===syslogMsgBuilder.mediate called=============
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:161,getSyslogMsgs. start
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:171,getSyslogMsgs. Reading msg 1 =============
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:229,readMsgLen. start
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.0=1
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.1=9
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:247,cBuffer.2=4
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:260,readMsgLen: strMsglen = '194'
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:283,readMsgLen: intMsglen = 194
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:299,readMsgContent. start. readLen=194
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:342,readMsgContent: msgContent=[<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] ], nlFound=false
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:195,getSyslogMsgs: no NL found: stop
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:85,Found 1 message(s). Delete existing payload
> 2013-09-05T14:14:53,DEBUG,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:105,syslogMsg.1='<170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] '
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:113,xmlStr.1='<text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text>'
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:132,Made 1 message(s)
> 2013-09-05T14:14:53,INFO ,mdzhsfesb01,MSGMED,syslogMsgBuilder.java:136,SOAP.Envelope: <?xml version='1.0' encoding='utf-8'?><soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"><soapenv:Body><text xmlns="http://ws.apache.org/commons/ns/payload"><170>1 2013-09-05T12:14:53.596Z N30094 DaMon - - [ignore_330660128@7410@7410 type="MONITORING_DCS_QUEUESTAT" status="Open" summary="Summary" details="Details" system="DCS" server="svupdcsus25"] </text></soapenv:Body></soapenv:Envelope>
>
> synapse.xml
> -----------
> <definitions xmlns="http://ws.apache.org/ns/synapse">
> <sequence name="fault">
> <makefault>
> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
> <reason value="Mediation failed."/>
> </makefault>
> <send/>
> </sequence>
> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
> <in>
> <log level="simple"/>
> <send/>
> </in>
> <out>
> <send/>
> </out>
> </sequence>
>
> <proxy name="proxyTcp2Jms" transports="tcp">
> <target>
> <inSequence>
> <!-- Define TCP listener expects plain text (syslog) messages -->
> <property name="messageType" value="text/plain" scope="axis2"/>
> <property name="OUT_ONLY" value="true"/>
> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
> <log level="simple"/> <!-- simple,headers,full -->
>
> <!-- Split any multiple syslog messages for the iterator -->
> <!-- and return as XML with child messages -->
> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
> <!-- <log level="full"/> -->
> </class>
>
> <!-- <log level="full"/> -->
>
> <!-- Iterate over any multiple messages -->
> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
> <target>
> <sequence>
> <send>
> <!-- <log level="full"/> -->
> <endpoint>
> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
> </endpoint>
> </send>
> </sequence>
> </target>
> </iterate>
> </inSequence>
> <!-- OUT_ONLY=true
> <outSequence>
> <log level="simple" category="INFO" separator=","/>
> <send/>
> </outSequence>
> -->
> <faultSequence>
> <log level="simple" category="ERROR" separator=","/>
> </faultSequence>
> </target>
> <parameter name="transport.tcp.port">6060</parameter>
> <parameter name="transport.tcp.contentType">text/plain</parameter>
> <!-- <parameter name="transport.tcp.backlog">0</parameter> -->
> </proxy>
> </definitions>
>
>
> syslogMsgBuilder.java
> ---------------------
> /*
> * Filename: syslogMsgBuilder.java
> * Author: P.Wright
> * Date: 06.08.2013
> * Description: Java class for Synapse class mediator.
> * Class is embedded in Synapse, and called when a SYSLOG message is received.
> *
> * The SYSLOG message can contain 1-n syslog messages, each
> * separated by a NL, and the length of each message defined
> * at the start of it. This is called TCP-Framing "octet-counted".
> * See: http://www.rsyslog.com/doc/omfwd.html
> *
> * Example input syslog message containing 3 messages
> * 35 <1> This is dummy message number 1\n
> * 35 <2> This is dummy message number 2\n
> * 34 <3> This is dummy message number 3
> *
> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
> * <?xml version='1.0' encoding='utf-8'?>
> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
> * <soapenv:Body>
> * <text xmlns="http://ws.apache.org/commons/ns/payload">
> * 35 <1> This is dummy message number 1\n
> * 35 <2> This is dummy message number 2\n
> * 34 <3> This is dummy message number 3
> * </text>
> * </soapenv:Body>
> * </soapenv:Envelope>
> *
> * Actions as follows:
> * - Get the first element of the message body
> * - Get a reader to the content of the first element
> * - Read the msglen
> * - Read the message according to msglen
> * - If last character is NOT a '\n', stop reading messages
> * - If last character IS a '\n', keep going and read next msglen
> * - Replace any "<" with "<"
> * - Replace the payload with an XML content with 3 messages (see below)
> *
> * Output message as follows:
> * <?xml version='1.0' encoding='utf-8'?>
> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
> * <soapenv:Body>
> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
> * </soapenv:Body>
> * </soapenv:Envelope>
> *
> * Method Overview:
> * - mediate ("main" method)
> */
>
> package com.sixtelekurs.classMediators;
>
> import java.io.Reader;
> import java.util.Iterator;
> import java.util.ArrayList;
> import java.util.List;
> import org.apache.axiom.om.OMElement;
> import org.apache.axiom.om.util.AXIOMUtil;
> import org.apache.axiom.soap.SOAPBody;
> import org.apache.axiom.soap.SOAPEnvelope;
> import org.apache.commons.logging.Log;
> import org.apache.commons.logging.LogFactory;
> import org.apache.synapse.MessageContext;
> import org.apache.synapse.Mediator;
>
>
> public class syslogMsgBuilder implements Mediator
> {
> private static String className = "syslogMsgBuilder";
> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
> private String _desc = null;
> private String _uid = null;
>
> // ------------------------------------------------------------------------
> public boolean mediate(MessageContext mc)
> {
> String logMsg = "===" + className + ".mediate called=============";
> log.info(logMsg);
>
> // Get the message content as an array of strings
> List<String> listSyslogStr = getSyslogMsgs(mc.getEnvelope().getBody().getFirstElement());
>
> // Delete the existing payload
> logMsg = "Found " + listSyslogStr.size() + " message(s). Delete existing payload";
> log.info(logMsg);
> SOAPBody soapBody = mc.getEnvelope().getBody();
> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
> {
> itr.next();
> itr.remove();
> }
>
> // Traverse the string list, convert each to an XML element
> // and write into the SOAP body
> int msgCnt = 0;
> try
> {
> // Get the syslog msg
> for (String msg : listSyslogStr)
> {
> // Replace all "<" with "<" for XML
> String syslogMsg = msg.replaceAll("<", "<");
> int jj = msgCnt+1;
> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
> log.debug(logMsg);
> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
> xmlStr += syslogMsg;
> xmlStr += "</text>";
> msgCnt++;
>
> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
> log.info(logMsg);
> soapBody.addChild(xmlElem);
> }
> }
> catch (Exception e)
> {
> logMsg = "ERROR: " + e;
> log.error(logMsg);
> return(false);
> }
>
> if (msgCnt == 0)
> {
> logMsg = "ERROR: no messages processed";
> log.error(logMsg);
> return(false);
> }
>
> logMsg = "Made " + msgCnt + " message(s)";
> log.info(logMsg);
>
> SOAPEnvelope envelope = mc.getEnvelope();
> logMsg = "SOAP.Envelope: " + envelope;
> log.info(logMsg);
>
> return(true);
> }
>
> class ReturnValues
> {
> String msgString;
> boolean foundNewline;
> ReturnValues(final String msgString, boolean foundNewline)
> {
> this.msgString = msgString;
> this.foundNewline = foundNewline;
> }
> }
>
> // getSyslogMsgs: traverses the input message
> // and returns any syslog messages as an array
> // of strings in XML form
> private List<String> getSyslogMsgs(OMElement omElem)
> {
> String funcname = "getSyslogMsgs";
> String logMsg = "";
>
> logMsg = funcname + ". start";
> log.debug(logMsg);
>
> List<String> syslogMsgs = new ArrayList<String>();
> Reader reader = omElem.getTextAsStream(false); // pass cache=false
>
> int msgCnt = 0;
> boolean keepGoing = true;
> while (keepGoing)
> {
> logMsg = funcname + ". Reading msg " + (msgCnt+1) + " =============";
> log.info(logMsg);
>
> // Get the msglen
> int msgLen = readMsgLen(reader);
> if (msgLen <= 0)
> {
> keepGoing = false;
> }
> else
> {
> // Get the message
> ReturnValues retvals = readMsgContent(reader, msgLen);
> if (retvals.msgString.isEmpty())
> {
> keepGoing = false;
> }
> else
> {
> syslogMsgs.add(retvals.msgString);
>
> // Read any trailing NL
> if (!retvals.foundNewline)
> {
> logMsg = funcname + ": no NL found: stop";
> log.debug(logMsg);
> keepGoing = false;
> }
> }
> }
> msgCnt++;
> }
>
> // Close the reader
> /*********************
> try
> {
> reader.close();
> }
> catch (Exception e)
> {
> logMsg = funcname + ": reader.close failed. " + e;
> log.error(logMsg);
> }
> *********************/
> return(syslogMsgs);
> }
>
> // Get the msglen of the syslog message
> // Read characters till a space and convert to an integer
> // Return the int msglen
> private int readMsgLen(Reader reader)
> {
> String funcname = "readMsgLen";
> int intMsglen = 0;
> String logMsg = "";
> String strMsglen = "";
>
> logMsg = funcname + ". start";
> log.debug(logMsg);
>
> try
> {
> char[] cBuffer = new char[10];
> int cCnt=0;
> boolean keepGoing = true;
> while (keepGoing)
> {
> // Read one character at a time into the char array
> int rc = reader.read(cBuffer, cCnt, 1);
> if (rc <= 0)
> {
> keepGoing = false;
> }
> else if (cBuffer[cCnt] != ' ')
> {
> logMsg = "cBuffer." + cCnt + "=" + cBuffer[cCnt];
> log.debug(logMsg);
> cCnt++;
> }
> else
> {
> keepGoing = false;
> }
> }
> if (cCnt > 0)
> {
> // Convert the characters read into a string
> strMsglen = new String(cBuffer, 0, cCnt);
> logMsg = funcname + ": strMsglen = '" + strMsglen + "'";
> log.debug(logMsg);
> }
> }
> catch (Exception e)
> {
> logMsg = funcname + ": error reading msglen" + e;
> log.error(logMsg);
> }
>
> if (strMsglen != "")
> {
> // Convert the (string)msglen to an (int)msglen
> try
> {
> intMsglen = Integer.parseInt(strMsglen);
> }
> catch (Exception e)
> {
> logMsg = funcname + ": error converting string '" + strMsglen + "' to integer. " + e;
> log.error(logMsg);
> }
> }
> logMsg = funcname + ": intMsglen = " + intMsglen;
> log.debug(logMsg);
> return(intMsglen);
> }
>
> // readMsgContent --------------------------------------------
> private ReturnValues readMsgContent(Reader reader, int msgLen)
> {
> String funcname = "readMsgContent";
> String msgContent = "";
> boolean nlFound = false;
> String logMsg = "";
> int maxlen = 2048;
> char[] cBuffer = new char[maxlen];
> int readLen = (msgLen < (maxlen-1) ? msgLen:(maxlen-1));
>
> logMsg = funcname + ". start. readLen=" + readLen;
> log.debug(logMsg);
>
> if (msgLen > maxlen-1)
> {
> logMsg = "message too long for buffer (" + msgLen + "/" + (maxlen-1);
> log.error(logMsg);
> ReturnValues retvals = new ReturnValues("", false);
> return(retvals);
> }
>
> try
> {
> int rc = reader.read(cBuffer, 0, readLen);
> if (rc <= 0)
> {
> logMsg = funcname + ": error reading msg: " + rc;
> log.error(logMsg);
> }
> else
> {
> if (rc < readLen)
> {
> logMsg = "read: IST=" + rc + ", SOLL=" + readLen;
> log.warn(logMsg);
> }
>
> // Ignore any trailing NL
> if (cBuffer[rc-1] == '\n')
> {
> nlFound = true;
> logMsg = "NL read at end of message - ignore it";
> log.debug(logMsg);
> rc--;
> }
> msgContent = new String(cBuffer, 0, rc);
> }
> }
> catch (Exception e)
> {
> logMsg = funcname + ": exception reading msg" + e;
> log.error(logMsg);
> }
> logMsg = funcname + ": msgContent=[" + msgContent + "], nlFound=" + nlFound;
> log.debug(logMsg);
>
> ReturnValues retvals = new ReturnValues(msgContent, nlFound);
> return(retvals);
> }
>
> // SET methods --------------------------------
> public String getDescription()
> {
> String msg = className + ".desc";
> return(msg);
> }
> public String getType()
> {
> String msg = className + ".type";
> return(msg);
> }
> public int getTraceState()
> {
> return(0);
> }
>
> // SET methods --------------------------------
> public void setTraceState(int traceState)
> {
> traceState = 0;
> }
> public void setDescription(String desc)
> {
> _desc = desc;
> }
> public void setUid(String uid)
> {
> _uid = uid;
> }
> }
>
> -----Original Message-----
> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
> Sent: Freitag, 30. August 2013 19:11
> To: user@synapse.apache.org
> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>
>
> On Aug 30, 2013, at 5:40 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>
>> Hi Hiranja,
>>
>> I've done a series of tests that appear to indicate the problem
>> is in the TCP transport.
>
> It's not really a problem in the tcp transport. The issue is exactly what I mentioned in my previous mail. That is, Synapse tries to read the input stream all the way to the end-of-stream, which causes it to get blocked. In addition to your custom mediator, other mediators like <lov level="full"/> also cause the same behavior. Basically any mediator that accesses the full message payload can cause this problem.
>
> The way I see it, you have only three options at this point:
>
> 1. Change your class mediator to not read the message payload to the end-of-stream. But you still have to get rid of any other mediators that may cause the message to be fully serialized (e.g. <log level="full"/>). Therefore this may not be very practical.
> 2. Write a custom message builder instead of a class mediator. This is probably the most architecturally sound approach. However, you must make sure that the builder doesn't read the input stream all the way to the end-of-stream. That is, it should read the message up to a known delimiter, or read a known number of bytes. Otherwise you will have the same problem again.
> 3. Use HTTP to receive the messages (HTTP protocol handles this type of issues using the Content-Length header or the chunked encoding system).
>
>> Here are the steps I did and the results I saw:
>> Important: "still the same" means:
>> - while the tcp connection is open, there is no reaction from synapse (no log messages etc)
>> - only AFTER the connection to the TCP listener has been closed do log messages appear and the syslog messages are processed
>>
>> 1. Rewrote my class mediator to use BufferedReader.readLine() -> still the same
>> 2. Rewrote my class mediator to simply replace the payload with fixed messages WITHOUT even reading the existing payload -> still the same
>> 3. Commented out my class mediator in synapse.xml -> still the same
>
> All the above 3 are mainly due to the other serializing mediators in the configuration (e.g. log). Also FYI, changing to the BufferedRead.readLine() doesn't make any difference. That is also a blocking I/O call.
>
>> 4. Added transport.tcp.port to the axis2.xml (as well as being in proxyTcp2Jms in synapse.xml) -> error at synapse startup "Error while starting the TCP endpoint. Address already in use"
>> 5. As above, but removed the port from proxyTcp2Jms in synapse.xml -> error at synapse startup "Service doesn't have configuration information for transport tcp"
>> 6. Defined transport.tcp.port in both files, but with different port numbers (6060 in synapse.xml, 6061 in axis2.xml) -> synapse starts up OK (with two logfile entries):
>> - TCPServer.java:76,TCP server started on port : 6061
>> - TCPServer.java:76,TCP server started on port : 6060
>> -> when I send a syslog message to port 6060 -> still the same
>> -> when I send a syslog message to port 6061 -> error written to logfile immediately: AxisEngine.java:219,The service cannot be found for the endpoint reference (EPR)
>> It appears that the TCPserver configured in synapse.xml is blocking until the client closes the connection,
>> and the TCPListener configured in axis2.xml is not blocking, but cannot find the proxy definition.
>
> That's not how it works. Messages sent to port 6061 must be dispatched by looking at the first element of the SOAP body. In your case it's just a generic <text/> element. Therefore Synapse fails to dispatch the message to any proxy service. This happens before any of the mediators are invoked. Therefore Synapse doesn't get to access the full message payload in this case. Hence it fails early in the pipeline, before Synapse gets blocked on I/O.
>
> Thanks,
> Hiranya
>
>>
>> The tcp jar file I'm using is this one:
>> http://maven.wso2.org/nexus/content/groups/wso2-public/org/apache/axis2/axis2-transport-tcp/1.1.0-SNAPSHOT/axis2-transport-tcp-1.1.0-SNAPSHOT.jar
>> but today also tried this one but saw no difference:
>> http://dist.wso2.org/maven2//org/apache/axis2/axis2-transport-tcp/1.1.0-wso2v5/axis2-transport-tcp-1.1.0-wso2v5.jar
>>
>> Thanks,
>> Peter
>>
>> -----Original Message-----
>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>> Sent: Montag, 26. August 2013 19:44
>> To: user@synapse.apache.org
>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>
>> Hi,
>>
>> On Aug 26, 2013, at 4:48 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>
>>> Hi Hiranja,
>>>
>>> No problem. Thanks for the support.
>>>
>>> Peter
>>>
>>> ===============================================================
>>>
>>> /*
>>> * Filename: syslogMsgBuilder.java
>>> * Author: P.Wright
>>> * Date: 06.08.2013
>>> * Description: Java class for Synapse class mediator.
>>> * Class is embedded in Synapse, and called when a SYSLOG message is received.
>>> *
>>> * The SYSLOG message can contain 1-n syslog messages, each
>>> * separated by a NL, and the length of each message defined
>>> * at the start of it. This is called TCP-Framing "octet-counted".
>>> * See: http://www.rsyslog.com/doc/omfwd.html
>>> *
>>> * Example input syslog message containing 3 messages
>>> * 34 <1> This is dummy message number 1
>>> * 34 <2> This is dummy message number 2
>>> * 34 <3> This is dummy message number 3
>>> *
>>> * Example Synapse-SOAP message with 3 syslog messages in the XML payload:
>>> * <?xml version='1.0' encoding='utf-8'?>
>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>> * <soapenv:Body>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload">
>>> * 34 <1> This is dummy message number 1
>>> * 34 <2> This is dummy message number 2
>>> * 34 <3> This is dummy message number 3
>>> * </text>
>>> * </soapenv:Body>
>>> * </soapenv:Envelope>
>>> *
>>> * Actions as follows:
>>> * - Get the payload
>>> * - Split the payload on NL
>>> * - Remove the msglen at the start of each message
>>> * - Replace any "<" with "<"
>>> * - Replace the payload with an XML content with 3 messages (see below)
>>> *
>>> * Output message as follows:
>>> * <?xml version='1.0' encoding='utf-8'?>
>>> * <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
>>> * <soapenv:Body>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><1>This is dummy message number 1</text>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><2>This is dummy message number 2</text>
>>> * <text xmlns="http://ws.apache.org/commons/ns/payload"><3>This is dummy message number 3</text>
>>> * </soapenv:Body>
>>> * </soapenv:Envelope>
>>> *
>>> * Method Overview:
>>> * - mediate ("main" method)
>>> */
>>> package com.sixtelekurs.classMediators;
>>>
>>> import java.util.Iterator;
>>> import org.apache.axiom.om.OMElement;
>>> import org.apache.axiom.om.util.AXIOMUtil;
>>> import org.apache.axiom.soap.SOAPBody;
>>> import org.apache.axiom.soap.SOAPEnvelope;
>>> import org.apache.commons.logging.Log;
>>> import org.apache.commons.logging.LogFactory;
>>> import org.apache.synapse.MessageContext;
>>> import org.apache.synapse.Mediator;
>>>
>>>
>>> public class syslogMsgBuilder implements Mediator
>>> {
>>> private static String className = "syslogMsgBuilder";
>>> private static final Log log = LogFactory.getLog(syslogMsgBuilder.class);
>>> private String _desc = null;
>>> private String _uid = null;
>>>
>>> // ------------------------------------------------------------------------
>>> public boolean mediate(MessageContext mc)
>>> {
>>> String logMsg = "===" + className + ".mediate called=============";
>>> log.debug(logMsg);
>>>
>>> // Get the message content and process it
>>> String syslogStr = mc.getEnvelope().getBody().getFirstElement().getText();
>>
>> I believe this line if the issue. The call to getText() on the OMSourcedElement probably causes it to read the entire input stream until end-of-stream is encountered. But end-of-stream does not occur until the sender closes its output stream at the remote end.
>>
>> I think a proper solution should get the InputStream from the OMSourcedElement and try to read from it until a known number of bytes have been read or a particular delimiter character is encountered. In a case like HTTP, the HTTP message usually have the content-length header or a special delimiter sequence (chunk delimiter) which allows the server to know how much data to read from the input stream without reading all the way to the end-of-stream. In case of TCP transport, there are no special headers to look at, so you must rely on the information available on the message payload it self.
>>
>> That's just my take on it. May be there's a better/easier way to solve this problem at Axiom level (perhaps a special method in the OMSourcedElement API?). May be Andreas can shed some light on the matter.
>>
>> Thanks,
>> Hiranya
>>
>>> logMsg = "syslogStr='" + syslogStr + "'";
>>> log.debug(logMsg);
>>>
>>> // Delete the existing payload
>>> logMsg = "Deleting existing payload";
>>> log.debug(logMsg);
>>> SOAPBody soapBody = mc.getEnvelope().getBody();
>>> for (Iterator itr = soapBody.getChildElements(); itr.hasNext();)
>>> {
>>> itr.next();
>>> itr.remove();
>>> }
>>>
>>> // Split the strings into N syslog messages, convert each to
>>> // an OMElement, and append to the XML-Body
>>> int msgCnt = 0;
>>> try
>>> {
>>> // Split the string on NL
>>> String[] lines = syslogStr.split("\\n");
>>> logMsg = "syslog.input.cnt(NL)=" + lines.length;
>>> log.debug(logMsg);
>>> for (int ii=0; ii < lines.length; ii++)
>>> {
>>> // Now split on space (once only) to ignore the msglen
>>> String[] msg = lines[ii].split(" ", 2);
>>> if (msg.length == 2)
>>> {
>>> // Replace all "<" with "<" for XML
>>> String syslogMsg = msg[1].replaceAll("<", "<");
>>> int jj = ii+1;
>>> logMsg = "syslogMsg." + jj + "='" + syslogMsg + "'";
>>> log.debug(logMsg);
>>> String xmlStr = "<text xmlns=\"http://ws.apache.org/commons/ns/payload\">";
>>> xmlStr += syslogMsg;
>>> xmlStr += "</text>";
>>> msgCnt++;
>>>
>>> // Now add elem
>>> OMElement xmlElem = AXIOMUtil.stringToOM(xmlStr);
>>> logMsg = "xmlStr." + jj + "='" + xmlStr + "'";
>>> log.debug(logMsg);
>>> soapBody.addChild(xmlElem);
>>> }
>>> else
>>> {
>>> logMsg = "msg could not be split correctly [" + msg + "]";
>>> log.error(logMsg);
>>> }
>>> }
>>> }
>>> catch (Exception e)
>>> {
>>> logMsg = "ERROR: " + e;
>>> log.error(logMsg);
>>> return(false);
>>> }
>>>
>>> if (msgCnt == 0)
>>> {
>>> logMsg = "ERROR: no messages processed";
>>> log.error(logMsg);
>>> return(false);
>>> }
>>>
>>> logMsg = "Made " + msgCnt + " messages";
>>> log.info(logMsg);
>>>
>>> SOAPEnvelope envelope = mc.getEnvelope();
>>> logMsg = "SOAP.Envelope: " + envelope;
>>> log.debug(logMsg);
>>>
>>> return(true);
>>> }
>>>
>>> // SET methods --------------------------------
>>> public String getDescription()
>>> {
>>> String msg = className + ".desc";
>>> return(msg);
>>> }
>>> public String getType()
>>> {
>>> String msg = className + ".type";
>>> return(msg);
>>> }
>>> public int getTraceState()
>>> {
>>> return(0);
>>> }
>>>
>>> // SET methods --------------------------------
>>> public void setTraceState(int traceState)
>>> {
>>> traceState = 0;
>>> }
>>> public void setDescription(String desc)
>>> {
>>> _desc = desc;
>>> }
>>> public void setUid(String uid)
>>> {
>>> _uid = uid;
>>> }
>>> }
>>>
>>> -----Original Message-----
>>> From: Hiranya Jayathilaka [mailto:hiranya911@gmail.com]
>>> Sent: Freitag, 23. August 2013 08:30
>>> To: user@synapse.apache.org
>>> Subject: Re: Synapse: TCP syslog to JMS proxy: messages only processed in synapse after client TCP connection is closed
>>>
>>> It sounds like something is trying to read from the input stream until the end of stream is encountered. This may be the plain text builder or your custom mediator. Can you share your custom mediator source code so we can take a look?
>>>
>>> Thanks,
>>> Hiranya
>>>
>>> On Aug 22, 2013, at 8:27 AM, "Wright, Peter" <Pe...@six-group.com> wrote:
>>>
>>>> Hi,
>>>>
>>>> I have my TCP Syslog --> JMS proxy up and running (see previous emails), but now have encountered another problem.
>>>> Up till now I have simply tested by using netstat to send 1-n messages stored in a file to the TCP server.
>>>> In this case, the messages are being received and processed correctly by synapse (sent as JMS
>>>> messages to the JMS server). It seems that this is working correctly, because the connection to the
>>>> synapse TCP server is being closed by netstat after it sends the messages.
>>>>
>>>> In my real world case however, we have 1-n clients writing syslog messages to a
>>>> local syslog server. This syslog server has been configured to send the messages on
>>>> to the (remote) synapse TCP server. In this case the messages are NOT being received
>>>> and processed by synapse. Only after the syslog server closes it connection to the synapse
>>>> TCP server, are the messages being received and processed by synapse.
>>>>
>>>> Any ideas what the problem could be?
>>>> Attached again my synapse.xml config file.
>>>>
>>>> Thanks,
>>>> Peter
>>>>
>>>> ===============================================
>>>>
>>>> <definitions xmlns="http://ws.apache.org/ns/synapse">
>>>> <sequence name="fault">
>>>> <makefault>
>>>> <code xmlns:tns="http://www.w3.org/2003/05/soap-envelope" value="tns:Receiver"/>
>>>> <reason value="Mediation failed."/>
>>>> </makefault>
>>>> <send/>
>>>> </sequence>
>>>> <sequence xmlns="http://ws.apache.org/ns/synapse" name="main" onError="fault">
>>>> <in>
>>>> <log level="full"/>
>>>> <send/>
>>>> </in>
>>>> <out>
>>>> <send/>
>>>> </out>
>>>> </sequence>
>>>>
>>>> <proxy name="proxyTcp2Jms" transports="tcp">
>>>> <target>
>>>> <inSequence>
>>>> <!-- Define TCP listener expects plain text (syslog) messages -->
>>>> <property name="messageType" value="text/plain" scope="axis2"/>
>>>> <property name="OUT_ONLY" value="true"/>
>>>> <property name="TRANSPORT_HEADERS" scope="axis2" action="remove"/>
>>>> <log level="full"/>
>>>>
>>>> <!-- Split any multiple syslog messages for the iterator -->
>>>> <!-- and return as XML with child messages -->
>>>> <class name="com.sixtelekurs.classMediators.syslogMsgBuilder">
>>>> <log level="full"/>
>>>> </class>
>>>>
>>>> <log level="full"/>
>>>>
>>>> <!-- Iterate over any multiple messages -->
>>>> <iterate id="syslogInterator" preservePayload="false" sequential="true" xmlns:m0="http://ws.apache.org/commons/ns/payload" expression="//m0:text">
>>>> <target>
>>>> <sequence>
>>>> <send>
>>>> <endpoint>
>>>> <address uri="jms:/cn=sed.finesb.syslog?java.naming.factory.initial=com.sun.jndi.ldap.LdapCtxFactory&java.naming.provider.url=LDAP_URL&transport.jms.ConnectionFactoryJNDIName=MY_TCF&transport.jms.DestinationType=topic&java.naming.security.principal=MY_DN&java.naming.security.credentials=MY_PASSWD"/>
>>>> </endpoint>
>>>> </send>
>>>> </sequence>
>>>> </target>
>>>> </iterate>
>>>> </inSequence>
>>>> <outSequence/>
>>>> <faultSequence>
>>>> <log level="full" category="ERROR" separator=","/>
>>>> </faultSequence>
>>>> </target>
>>>> <parameter name="transport.tcp.port">6060</parameter>
>>>> <parameter name="transport.tcp.contentType">text/plain</parameter>
>>>> </proxy>
>>>>
>>>> </definitions>
>>>>
>>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>>> Thank you.
>>>
>>> --
>>> Hiranya Jayathilaka
>>> Mayhem Lab/RACE Lab;
>>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>>> Blog: http://techfeast-hiranya.blogspot.com
>>>
>>> The content of this e-mail is intended only for the confidential use of the person addressed.
>>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>>> Thank you.
>>
>> --
>> Hiranya Jayathilaka
>> Mayhem Lab/RACE Lab;
>> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
>> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
>> Blog: http://techfeast-hiranya.blogspot.com
>>
>> The content of this e-mail is intended only for the confidential use of the person addressed.
>> If you are not the intended recipient, please notify the sender and delete this email immediately.
>> Thank you.
>
> --
> Hiranya Jayathilaka
> Mayhem Lab/RACE Lab;
> Dept. of Computer Science, UCSB; http://cs.ucsb.edu
> E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
> Blog: http://techfeast-hiranya.blogspot.com
>
> The content of this e-mail is intended only for the confidential use of the person addressed.
> If you are not the intended recipient, please notify the sender and delete this email immediately.
> Thank you.
--
Hiranya Jayathilaka
Mayhem Lab/RACE Lab;
Dept. of Computer Science, UCSB; http://cs.ucsb.edu
E-mail: hiranya@cs.ucsb.edu; Mobile: +1 (805) 895-7443
Blog: http://techfeast-hiranya.blogspot.com