Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #585 from garyrussell/INT-2711

Browse Source 
* INT-2711:
  INT-2711 syslog Transformer
This commit is contained in:
Oleg Zhurakousky
2012-08-14 14:16:13 -04:00
parent ca7e78a2d2 f8782f3dc9
commit a544ef08b3
14 changed files with 596 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractTransformerParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,11 +16,11 @@
package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.transformer.MessageTransformingHandler;
import org.w3c.dom.Element;
/**
* @author Mark Fisher
@@ -30,7 +30,7 @@ public abstract class AbstractTransformerParser extends AbstractConsumerEndpoint
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.MessageTransformingHandler");
MessageTransformingHandler.class);
BeanDefinitionBuilder transformerBuilder =
BeanDefinitionBuilder.genericBeanDefinition(this.getTransformerClassName());
this.parseTransformer(element, parserContext, transformerBuilder);

3
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceHandler.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.config.xml;
/**
* Namespace handler for the integration namespace.
*
*
* @author Mark Fisher
* @author Marius Bogoevici
* @author Oleg Zhurakousky
@@ -51,6 +51,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan
registerBeanDefinitionParser("payload-serializing-transformer", new PayloadSerializingTransformerParser());
registerBeanDefinitionParser("payload-deserializing-transformer", new PayloadDeserializingTransformerParser());
registerBeanDefinitionParser("claim-check-in", new ClaimCheckInParser());
registerBeanDefinitionParser("syslog-to-map-transformer", new SyslogToMapTransformerParser());
registerBeanDefinitionParser("claim-check-out", new ClaimCheckOutParser());
registerBeanDefinitionParser("inbound-channel-adapter", new DefaultInboundChannelAdapterParser());
registerBeanDefinitionParser("resource-inbound-channel-adapter", new ResourceInboundChannelAdapterParser());

40
spring-integration-core/src/main/java/org/springframework/integration/config/xml/SyslogToMapTransformerParser.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config.xml;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.transformer.SyslogToMapTransformer;
import org.w3c.dom.Element;
/**
* @author Gary Russell
* @since 2.2
*
*/
public class SyslogToMapTransformerParser extends AbstractTransformerParser {
@Override
protected String getTransformerClassName() {
return SyslogToMapTransformer.class.getName();
}
@Override
protected void parseTransformer(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
// no attributes
}
}

145
spring-integration-core/src/main/java/org/springframework/integration/transformer/SyslogToMapTransformer.java Normal file
View File

@@ -0,0 +1,145 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.transformer;
import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.springframework.util.Assert;
/**
* Transforms a packet in Syslog (RFC5424) format to a Map.
* If the packet cannot be decoded, the entire packet
* is returned as a String under the key UNDECODED. If the date field can be
* parsed, it will be returned as a {@link Date} object; otherwise it is returned as a String.
*
* @author Gary Russell
* @since 2.2
*
*/
public class SyslogToMapTransformer extends AbstractPayloadTransformer<Object, Map<String, ?>> {
public static final String FACILITY = "FACILITY";
public static final String SEVERITY = "SEVERITY";
public static final String TIMESAMP = "TIMESTAMP";
public static final String HOST = "HOST";
public static final String TAG = "TAG";
public static final String MESSAGE = "MESSAGE";
public static final String UNDECODED = "UNDECODED";
private final Pattern pattern = Pattern.compile("<([^>]+)>(.{15}) ([^ ]+) ([^:]+): (.*)", Pattern.DOTALL);
private final SimpleDateFormat dateFormat = new SimpleDateFormat("MMM dd HH:mm:ss");
private Map<String, ?> transform(byte[] payloadBytes) {
String payload;
try {
payload = new String(payloadBytes, "UTF-8");
}
catch (UnsupportedEncodingException e) {
payload = new String(payloadBytes);
}
return transform(payload);
}
private Map<String, ?> transform(String payload) {
Map<String, Object> map = new HashMap<String, Object>();
Matcher matcher = pattern.matcher(payload);
if (matcher.matches()) {
try {
String facilityString = matcher.group(1);
int facility = Integer.parseInt(facilityString);
int severity = facility & 0x7;
facility = facility >> 3;
map.put(FACILITY, facility);
map.put(SEVERITY, severity);
String timestamp = matcher.group(2);
Date date;
try {
date = this.dateFormat.parse(timestamp);
Calendar calendar = Calendar.getInstance();
int year = calendar.get(Calendar.YEAR);
int month = calendar.get(Calendar.MONTH);
calendar.setTime(date);
/*
* syslog date doesn't include a year so we
* need to insert the current year - adjusted
* if necessary if close to midnight on Dec 31.
*/
if (month == 11 && calendar.get(Calendar.MONTH) == 0) {
calendar.set(Calendar.YEAR, year + 1);
}
else if (month == 0 && calendar.get(Calendar.MONTH) == 1) {
calendar.set(Calendar.YEAR, year - 1);
}
else {
calendar.set(Calendar.YEAR, year);
}
map.put(TIMESAMP, calendar.getTime());
}
catch (Exception e) {
/*
* If we can't parse the timestamp, return it as an
* unmodified String. (Postel's law).
*/
map.put(TIMESAMP, timestamp);
}
map.put(HOST, matcher.group(3));
map.put(TAG, matcher.group(4));
map.put(MESSAGE, matcher.group(5));
}
catch (Exception e) {
if (logger.isDebugEnabled()) {
logger.debug("Could not decode:" + payload, e);
}
map.clear();
map.put(UNDECODED, payload);
}
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Could not decode:" + payload);
}
map.put(UNDECODED, payload);
}
return map;
}
@Override
protected Map<String, ?> transformPayload(Object payload) throws Exception {
Assert.isTrue(payload instanceof byte[] || payload instanceof String,
"payload must be String or byte[]");
if (payload instanceof byte[]) {
return this.transform((byte[]) payload);
}
else if (payload instanceof String) {
return this.transform((String) payload);
}
return null;
}
}

11
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd
View File

@@ -2209,6 +2209,17 @@
</xsd:attribute>
</xsd:complexType>
<xsd:element name="syslog-to-map-transformer">
<xsd:annotation>
<xsd:documentation>
Defines a Transformer that converts an RFC5424 packet to a Map.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
</xsd:complexType>
</xsd:element>
<!-- Claim Check -->
<xsd:element name="claim-check-in" type="claimCheckInType">

13
spring-integration-core/src/test/java/org/springframework/integration/config/xml/SyslogTransformerParserTests-context.xml Normal file
View File

@@ -0,0 +1,13 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.2.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:syslog-to-map-transformer id="toMap" input-channel="toMapChannel" output-channel="out" />
<int:channel id="out">
<int:queue />
</int:channel>
</beans>

64
spring-integration-core/src/test/java/org/springframework/integration/config/xml/SyslogTransformerParserTests.java Normal file
View File

@@ -0,0 +1,64 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.config.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.Date;
import java.util.Map;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.transformer.SyslogToMapTransformer;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 2.2
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class SyslogTransformerParserTests {
@Autowired
private MessageChannel toMapChannel;
@Autowired
private PollableChannel out;
@Test
public void testMap() {
toMapChannel.send(new GenericMessage<String>("<157>JUL 26 22:08:35 WEBERN TESTING[70729]: TEST SYSLOG MESSAGE"));
Map<?, ?> map = (Map<?, ?>) out.receive(1000).getPayload();
assertNotNull(map);
assertEquals(6, map.size());
System.out.println(map);
assertEquals(19, map.get(SyslogToMapTransformer.FACILITY));
assertEquals(5, map.get(SyslogToMapTransformer.SEVERITY));
assertTrue(map.get(SyslogToMapTransformer.TIMESAMP) instanceof Date);
assertEquals("WEBERN", map.get(SyslogToMapTransformer.HOST));
assertEquals("TESTING[70729]", map.get(SyslogToMapTransformer.TAG));
assertEquals("TEST SYSLOG MESSAGE", map.get(SyslogToMapTransformer.MESSAGE));
}
}

67
spring-integration-core/src/test/java/org/springframework/integration/transformer/SysLogTransformerTests.java Normal file
View File

@@ -0,0 +1,67 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.transformer;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.Date;
import java.util.Map;
import org.junit.Test;
/**
* @author Gary Russell
* @since 2.2
*
*/
public class SysLogTransformerTests {
@Test
public void testMap() throws Exception {
SyslogToMapTransformer t = new SyslogToMapTransformer();
Map<String, ?> transformed = t.transformPayload(
"<158>JUL 26 22:08:35 WEBERN TESTING[70729]: TEST SYSLOG MESSAGE".getBytes());
assertEquals(6, transformed.size());
// System.out.println(transformed);
assertEquals(19, transformed.get(SyslogToMapTransformer.FACILITY));
assertEquals(6, transformed.get(SyslogToMapTransformer.SEVERITY));
assertTrue(transformed.get(SyslogToMapTransformer.TIMESAMP) instanceof Date);
assertEquals("WEBERN", transformed.get(SyslogToMapTransformer.HOST));
assertEquals("TESTING[70729]", transformed.get(SyslogToMapTransformer.TAG));
assertEquals("TEST SYSLOG MESSAGE", transformed.get(SyslogToMapTransformer.MESSAGE));
}
@Test
public void testBadPattern() throws Exception {
SyslogToMapTransformer t = new SyslogToMapTransformer();
String syslog = "&158>JUL 26 22:08:35 WEBERN TESTING[70729]: TEST SYSLOG MESSAGE";
Map<String, ?> transformed = t.transformPayload(
syslog.getBytes());
assertEquals(1, transformed.size());
assertEquals(syslog, transformed.get(SyslogToMapTransformer.UNDECODED));
}
@Test
public void testBadFacilitySeverity() throws Exception {
SyslogToMapTransformer t = new SyslogToMapTransformer();
String syslog = "<X58>JUL 26 22:08:35 WEBERN TESTING[70729]: TEST SYSLOG MESSAGE";
Map<String, ?> transformed = t.transformPayload(
syslog.getBytes());
assertEquals(1, transformed.size());
assertEquals(syslog, transformed.get(SyslogToMapTransformer.UNDECODED));
}
}

28
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayLfSerializer.java Normal file
View File

@@ -0,0 +1,28 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.serializer;
/**
* @author Gary Russell
* @since 2.2
*
*/
public class ByteArrayLfSerializer extends ByteArraySingleTerminatorSerializer {
public ByteArrayLfSerializer() {
super((byte) 0x0a);
}
}

82
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArraySingleTerminatorSerializer.java Normal file
View File

@@ -0,0 +1,82 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp.serializer;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
/**
* Reads data in an InputStream to a byte[]; data must be terminated by a single
* byte (not included in resulting byte[]).
* Writes a byte[] to an OutputStream and adds the terminator.
*
* @author Gary Russell
* @since 2.2
*/
public class ByteArraySingleTerminatorSerializer extends AbstractByteArraySerializer {
private final byte terminator;
public ByteArraySingleTerminatorSerializer(byte delimiter) {
this.terminator = delimiter;
}
/**
* Reads the data in the inputstream to a byte[]. Data must be terminated
* by a single byte. Throws a {@link SoftEndOfStreamException} if the stream
* is closed immediately after the terminator (i.e. no data is in the process of
* being read).
*/
public byte[] deserialize(InputStream inputStream) throws IOException {
byte[] buffer = new byte[this.maxMessageSize];
int n = 0;
int bite;
if (logger.isDebugEnabled()) {
logger.debug("Available to read:" + inputStream.available());
}
while (true) {
bite = inputStream.read();
// logger.debug("Read:" + (char) bite);
if (bite < 0 && n == 0) {
throw new SoftEndOfStreamException("Stream closed between payloads");
}
checkClosure(bite);
if (n > 0 && bite == terminator) {
break;
}
buffer[n++] = (byte) bite;
if (n >= this.maxMessageSize) {
throw new IOException("LF not found before max message length: "
+ this.maxMessageSize);
}
};
byte[] assembledData = new byte[n];
System.arraycopy(buffer, 0, assembledData, 0, n);
return assembledData;
}
/**
* Writes the byte[] to the stream and appends the terminator.
*/
public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
outputStream.write(bytes);
outputStream.write(terminator);
outputStream.flush();
}
}

39
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SyslogdTests-context.xml Normal file
View File

@@ -0,0 +1,39 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:stream="http://www.springframework.org/schema/integration/stream"
xmlns:ip="http://www.springframework.org/schema/integration/ip"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/stream
http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
http://www.springframework.org/schema/integration/ip
http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
<!-- rsyslog conf:
$ModLoad omfwd
*.* @@localhost:1514
-->
<ip:tcp-connection-factory id="syslogListener"
type="server"
port="1514"
deserializer="lfDeser" />
<ip:tcp-inbound-channel-adapter channel="syslogChannel"
connection-factory="syslogListener" />
<beans:bean id="lfDeser" class="org.springframework.integration.ip.tcp.serializer.ByteArrayLfSerializer" />
<syslog-to-map-transformer input-channel="syslogChannel" output-channel="out" />
<channel id="out" />
<stream:stdout-channel-adapter channel="out" append-newline="true"/>
</beans:beans>

35
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SyslogdTests.java Normal file
View File

@@ -0,0 +1,35 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.tcp;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author Gary Russell
* @since 2.2
*
*/
public class SyslogdTests {
public static void main(String[] args) throws Exception {
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("SyslogdTests-context.xml", SyslogdTests.class);
System.out.println("Hit enter to terminate");
System.in.read();
ctx.destroy();
}
}

32
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/SyslogdTests-context.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:stream="http://www.springframework.org/schema/integration/stream"
xmlns:ip="http://www.springframework.org/schema/integration/ip"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/stream
http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
http://www.springframework.org/schema/integration/ip
http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd">
<!-- rsyslog conf:
*.* @localhost:1513
-->
<ip:udp-inbound-channel-adapter id="udpReceiver"
channel="syslogChannel"
port="1513" />
<syslog-to-map-transformer input-channel="syslogChannel" output-channel="out" />
<channel id="out" />
<stream:stdout-channel-adapter channel="out" append-newline="true"/>
</beans:beans>

34
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/SyslogdTests.java Normal file
View File

@@ -0,0 +1,34 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.ip.udp;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author Gary Russell
* @since 2.2
*
*/
public class SyslogdTests {
public static void main(String[] args) throws Exception {
AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("SyslogdTests-context.xml", SyslogdTests.class);
System.out.println("Hit enter to terminate");
System.in.read();
ctx.destroy();
}
}