INT-3450 Support RFC 5424/6587 for SYSLOG

JIRA: https://jira.spring.io/browse/INT-3450

Use LinkedHashMap

- natural iteration order

INT-3450 Polishing - PR Comments

INT-3450: Polishing

spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java

@@ -59,6 +59,10 @@ public class DefaultMessageConverter implements MessageConverter, BeanFactoryAwa
 		this.asMap = asMap;
 	}
 
+	protected boolean asMap() {
+		return asMap;
+	}
+
 	@Override
 	public final void setBeanFactory(BeanFactory beanFactory) {
 		this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);

spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424MessageConverter.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2014 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.syslog;
+
+import java.util.Map;
+
+import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+/**
+ * {@link MessageConverter}; delegates to a {@link RFC5424SyslogParser} if
+ * necessary (TCP will have already done the syslog conversion because it needs
+ * to handle different message framing). Copies the resulting {@link Map} to
+ * the message headers if {@link #asMap()} is false.
+ *
+ * @author Gary Russell
+ * @since 4.1.1
+ */
+public class RFC5424MessageConverter extends DefaultMessageConverter {
+
+	private final RFC5424SyslogParser parser;
+
+	private String charset = "UTF-8";
+
+	/**
+	 * Construct an instance with a default {@link RFC5424SyslogParser}.
+	 */
+	public RFC5424MessageConverter() {
+		this(new RFC5424SyslogParser());
+	}
+
+	/**
+	 * Construct an instance with a non-standard parser.
+	 * @param parser the parser.
+	 */
+	public RFC5424MessageConverter(RFC5424SyslogParser parser) {
+		this.parser = parser;
+	}
+
+	/**
+	 * @param charset the charset to set
+	 */
+	protected void setCharset(String charset) {
+		this.charset = charset;
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public Message<?> fromSyslog(Message<?> message) throws Exception {
+		boolean isMap = message.getPayload() instanceof Map;
+		Map<String, ?> map;
+		Object originalContent;
+		if (!isMap) {
+			Assert.isInstanceOf(byte[].class, message.getPayload(), "Only byte[] and Map payloads are supported");
+			map = this.parser.parse(new String(((byte[]) message.getPayload()), this.charset), 0, false);
+			originalContent = message.getPayload();
+		}
+		else {
+			map = (Map<String, ?>) message.getPayload();
+			originalContent = map.get(SyslogHeaders.UNDECODED);
+			if (originalContent == null) {
+				originalContent = map;
+			}
+		}
+
+		AbstractIntegrationMessageBuilder<Object> builder = getMessageBuilderFactory().withPayload(
+				asMap() ? map : originalContent);
+		if (!asMap() && isMap) {
+			builder.copyHeaders(map);
+		}
+		return builder.build();
+	}
+
+}

spring-integration-syslog/src/main/java/org/springframework/integration/syslog/RFC5424SyslogParser.java

@@ -0,0 +1,386 @@
+/*
+ * Copyright 2014 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.syslog;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.util.Assert;
+
+/**
+ * Parse for RFC 5424 syslog messages; when used with TCP, requires the use
+ * of a {@code RFC6587SyslogDeserializer} which decodes the framing.
+ *
+ * @author Duncan McIntyre
+ * @author Gary Russell
+ * @since 1.4.1
+ *
+ */
+public class RFC5424SyslogParser {
+
+	protected static final char NILVALUE = '-';
+
+	protected static final char SPACE = ' ';
+
+	protected final boolean retainOriginal;
+
+
+	/**
+	 * Construct a default parser; do not retain the original message content unless there
+	 * is an error.
+	 */
+	public RFC5424SyslogParser() {
+		this(false);
+	}
+
+	/**
+	 * @param retainOriginal when true, include the original message content intact in the
+	 * map.
+	 */
+	public RFC5424SyslogParser(boolean retainOriginal) {
+		this.retainOriginal = retainOriginal;
+	}
+
+	public Map<String, ?> parse(String line, int octetCount, boolean shortRead) {
+
+		Map<String, Object> map = new LinkedHashMap<String, Object>();
+
+		Reader r = new Reader(line);
+
+		try {
+			if (shortRead) {
+				int n = line.length() - 1;
+				while (n >= 0 && line.charAt(n) == 0x00) {
+					n--;
+				}
+				line = line.substring(0, n);
+				throw new IllegalStateException("Insufficient data; expected " + octetCount + " got " + (n + 1));
+			}
+			r.expect('<');
+			int pri = r.readInt();
+			r.expect('>');
+
+			int version = r.readInt();
+			r.expect(SPACE);
+
+			Object timestamp = getTimestamp(r);
+
+			String host = r.getIdentifier();
+			String app = r.getIdentifier();
+			String procId = r.getIdentifier();
+			String msgId = r.getIdentifier();
+
+			Object structuredData = getStructuredData(r);
+
+			String message;
+			if(r.is(SPACE)) {
+				r.getc();
+				message = r.rest();
+			}
+			else {
+				message = "";
+			}
+
+			int severity = pri & 0x7;
+			int facility = pri >> 3;
+			map.put(SyslogHeaders.FACILITY, facility);
+
+			map.put(SyslogHeaders.SEVERITY, severity);
+			map.put(SyslogHeaders.SEVERITY_TEXT, Severity.parseInt(severity).label());
+
+			if (timestamp != null) {
+				map.put(SyslogHeaders.TIMESTAMP, timestamp);
+			}
+
+			if (host != null) {
+				map.put(SyslogHeaders.HOST, host);
+			}
+			if (app != null) {
+				map.put(SyslogHeaders.APP_NAME, app);
+			}
+			if (procId != null) {
+				map.put(SyslogHeaders.PROCID, procId);
+			}
+			if (msgId != null) {
+				map.put(SyslogHeaders.MSGID, msgId);
+			}
+			map.put(SyslogHeaders.VERSION, version);
+
+			if (structuredData != null) {
+				map.put(SyslogHeaders.STRUCTURED_DATA, structuredData);
+			}
+
+			map.put(SyslogHeaders.MESSAGE, message);
+			map.put(SyslogHeaders.DECODE_ERRORS, "false");
+
+			if (this.retainOriginal) {
+				map.put(SyslogHeaders.UNDECODED, line);
+			}
+		}
+		catch(IllegalStateException e) {
+			map.put(SyslogHeaders.DECODE_ERRORS, "true");
+			map.put(SyslogHeaders.ERRORS, e.getMessage());
+			map.put(SyslogHeaders.UNDECODED, line);
+		}
+		catch(StringIndexOutOfBoundsException sob) {
+			map.put(SyslogHeaders.DECODE_ERRORS, "true");
+			map.put(SyslogHeaders.ERRORS, "Unexpected end of message: " + sob.getMessage());
+			map.put(SyslogHeaders.UNDECODED, line);
+		}
+		return map;
+	}
+
+	/**
+	 * Default implementation returns the date as a String (if present).
+	 * @param r the reader.
+	 * @return the timestamp.
+	 */
+	protected Object getTimestamp(Reader r) {
+
+		int c = r.getc();
+
+		if(c == NILVALUE) {
+			return null;
+		}
+
+		if(!Character.isDigit(c)) {
+			throw new IllegalStateException("Year expected @" + r.idx);
+		}
+
+		StringBuilder dateBuilder = new StringBuilder();
+		dateBuilder.append((char) c);
+		while ((c = r.getc()) != SPACE) {
+			dateBuilder.append((char) c);
+		}
+
+		return dateBuilder.toString();
+	}
+
+	private Object getStructuredData(Reader r) {
+		if(r.is(NILVALUE)) {
+			r.getc();
+			return null;
+		}
+		return parseStructuredDataElements(r);
+	}
+
+	/**
+	 * Default implementation returns a list of structured data elements with
+	 * no internal parsing.
+	 * @param r the reader.
+	 * @return the structured data.
+	 */
+	protected Object parseStructuredDataElements(Reader r) {
+		List<String> fragments = new ArrayList<String>();
+		while (r.is('[')) {
+			r.mark();
+			r.skipTo(']');
+			fragments.add(r.getMarkedSegment());
+		}
+		return fragments;
+	}
+
+	protected class Reader {
+
+		private final String line;
+
+		public int idx;
+
+		private int mark;
+
+		public Reader(String l) {
+			line = l;
+		}
+
+		public void mark() {
+			this.mark = this.idx;
+		}
+
+		public String getMarkedSegment() {
+			Assert.state(this.mark <= this.idx, "mark is greater than this.idx");
+			return this.line.substring(mark, this.idx);
+		}
+
+		public int current() {
+			return line.charAt(this.idx);
+		}
+
+		public int prev() {
+			return line.charAt(this.idx - 1);
+		}
+
+		public int getc() {
+			return line.charAt(this.idx++);
+		}
+
+		public int peek() {
+			return line.charAt(this.idx + 1);
+		}
+
+		public void ungetc() {
+			this.idx--;
+		}
+
+		public int getInt() {
+			int c = getc();
+			if (!Character.isDigit(c)) {
+				ungetc();
+				return -1;
+			}
+
+			return c - '0';
+		}
+
+		/**
+		 * Read characters building an int until a non-digit is found
+		 * @return int
+		 */
+		public int readInt() {
+			int val = 0;
+			while (isDigit()) {
+				val = (val * 10) + getInt();
+			}
+			return val;
+		}
+
+		public double readFraction() {
+			int val = 0;
+			int order = 0;
+			while (isDigit()) {
+				val = (val * 10) + getInt();
+				order *= 10;
+			}
+			return (double) val / order;
+
+		}
+
+		public boolean is(char c) {
+			return line.charAt(this.idx) == c;
+		}
+
+		public boolean was(char c) {
+			return line.charAt(this.idx - 1) == c;
+		}
+
+		public boolean isDigit() {
+			return Character.isDigit(line.charAt(this.idx));
+		}
+
+		public void expect(char c) {
+			if (line.charAt(this.idx++) != c) {
+				throw new IllegalStateException("Expected '" + c + "' @" + this.idx);
+			}
+		}
+
+		public void skipTo(char searchChar) {
+			while (!is(searchChar) || was('\\')) {
+				getc();
+			}
+			getc();
+		}
+
+		public String rest() {
+			return line.substring(this.idx);
+		}
+
+		public String getIdentifier() {
+			StringBuilder sb = new StringBuilder();
+			int c;
+			while (true) {
+				c = getc();
+				if (c >= 33 && c <= 127) {
+					sb.append((char) c);
+				}
+				else {
+					break;
+				}
+			}
+			return sb.toString();
+		}
+
+	}
+
+	protected enum Severity {
+
+		DEBUG(7, "DEBUG"),
+
+		INFO(6, "INFO"),
+
+		NOTICE(5, "NOTICE"),
+
+		WARN(4, "WARN"),
+
+		ERROR(3, "ERRORS"),
+
+		CRITICAL(2, "CRITICAL"),
+
+		ALERT(1, "ALERT"),
+
+		EMERGENCY(0, "EMERGENCY"),
+
+		UNDEFINED(-1, "UNDEFINED");
+
+		private final int level;
+
+		private final String label;
+
+		private Severity(int level, String label) {
+			this.level = level;
+			this.label = label;
+		}
+
+		public int level() {
+			return level;
+		}
+
+		public String label() {
+			return label;
+		}
+
+		public static Severity parseInt(int syslogSeverity) {
+			if(syslogSeverity == 7) {
+				return DEBUG;
+			}
+			if(syslogSeverity == 6) {
+				return INFO;
+			}
+			if(syslogSeverity == 5) {
+				return NOTICE;
+			}
+			if(syslogSeverity == 4) {
+				return WARN;
+			}
+			if(syslogSeverity == 3) {
+				return ERROR;
+			}
+			if(syslogSeverity == 2) {
+				return CRITICAL;
+			}
+			if(syslogSeverity == 1) {
+				return ALERT;
+			}
+			if(syslogSeverity == 0) {
+				return EMERGENCY;
+			}
+			return UNDEFINED;
+		}
+
+	}
+
+
+}

spring-integration-syslog/src/main/java/org/springframework/integration/syslog/SyslogHeaders.java

@@ -38,4 +38,33 @@ public class SyslogHeaders {
 
 	public static final String TAG = PREFIX + SyslogToMapTransformer.TAG;
 
+	//// RFC 5424 Parser:
+
+	public static final String MESSAGE = PREFIX + "MESSAGE";
+
+	public static final String APP_NAME = PREFIX + "APP_NAME" ;
+
+	public static final String PROCID = PREFIX + "PROCID" ;
+
+	public static final String MSGID = PREFIX + "MSGID" ;
+
+	public static final String VERSION = PREFIX + "VERSION" ;
+
+	public static final String STRUCTURED_DATA = PREFIX + "STRUCTURED_DATA" ;
+
+	// Text versions of syslog numeric values
+	public static final String SEVERITY_TEXT = PREFIX + "SEVERITY_TEXT" ;
+
+	// Additional fields
+	public static final String SOURCE_TYPE = PREFIX + "SOURCE_TYPE" ;
+
+	public static final String SOURCE = PREFIX + "SOURCE" ;
+
+	// full line when parse errors or retained original
+	public static final String UNDECODED = PREFIX + "UNDECODED" ;
+
+	public static final String DECODE_ERRORS = PREFIX + "DECODE_ERRORS" ;
+
+	public static final String ERRORS = PREFIX + "ERRORS";
+
 }

spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/RFC6587SyslogDeserializer.java

@@ -0,0 +1,119 @@
+/*
+ * Copyright 2014 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.syslog.inbound;
+
+import java.io.DataInputStream;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+
+import org.springframework.core.serializer.Deserializer;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayLfSerializer;
+import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException;
+import org.springframework.integration.syslog.RFC5424SyslogParser;
+import org.springframework.util.Assert;
+
+/**
+ * RFC5424/6587 Deserializer. Implemented as a {@link Deserializer} instead of a
+ * transformer because we may receive a mixture of octet counting and non-transparent
+ * framing - see RFC 6587.
+ *
+ * @author Duncan McIntyre
+ * @author Gary Russell
+ * @since 4.1.1
+ *
+ */
+public class RFC6587SyslogDeserializer implements Deserializer<Map<String, ?>> {
+
+	private final Deserializer<byte[]> delimitedDeserializer;
+
+	private RFC5424SyslogParser parser = new RFC5424SyslogParser();
+
+	/**
+	 * Construct an instance using a {@link ByteArrayLfSerializer} for
+	 * non-transparent frames.
+	 */
+	public RFC6587SyslogDeserializer() {
+		this.delimitedDeserializer = new ByteArrayLfSerializer();
+	}
+
+	/**
+	 * Construct an instance using the specified {@link Deserializer} for
+	 * non-transparent frames.
+	 * @param delimitedDeserializer the Deserializer.
+	 */
+	public RFC6587SyslogDeserializer(Deserializer<byte[]> delimitedDeserializer) {
+		this.delimitedDeserializer = delimitedDeserializer;
+	}
+
+	/**
+	 * @param parser the parser to set
+	 */
+	public void setParser(RFC5424SyslogParser parser) {
+		this.parser = parser;
+	}
+
+	@Override
+	public Map<String, ?> deserialize(InputStream inputStream) throws IOException {
+		DataInputStream stream = new DataInputStream(inputStream);
+		String line;
+		int octetCount = 0;
+		boolean shortRead = false;
+		int peek = stream.read();
+		if (isDigit(peek)) {
+			octetCount = calculateLength(stream, peek);
+			Assert.state(octetCount > 0, "Expected length > 0");
+			byte[] bytes = new byte[octetCount];
+			try {
+				stream.readFully(bytes);
+			}
+			catch (EOFException e) {
+				shortRead = true;
+			}
+			line = new String(bytes, getCharset());
+		}
+		else if (peek == '<') {
+			byte[] bytes = this.delimitedDeserializer.deserialize(inputStream);
+			line = "<" + new String(bytes, getCharset());
+		}
+		else if (peek < 0) {
+			throw new SoftEndOfStreamException();
+		}
+		else {
+			throw new IllegalStateException("Expected a digit or '<', got 0x" + Integer.toHexString(peek));
+		}
+		return parser.parse(line, octetCount, shortRead);
+	}
+
+	private boolean isDigit(int peek) {
+		return peek >= 0x30 && peek <= 0x39;
+	}
+
+	private int calculateLength(DataInputStream stream, int peek) throws IOException {
+		int length = peek & 0xf;
+		int c;
+		while (isDigit((c = stream.read()))) {
+			length = length*10 + (c & 0xf);
+		}
+		return length;
+	}
+
+	protected String getCharset() {
+		return "UTF-8";
+	}
+
+}

spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests-context.xml

@@ -54,9 +54,18 @@
 		connection-factory="cf"
 		auto-startup="false"
 		phase="123"
-		converter="converter"
+		converter="rfc5424"
 		send-timeout="456"
 		error-channel="errors" />
 
-	<int-ip:tcp-connection-factory id="cf" type="server" port="1514" />
+	<int-ip:tcp-connection-factory id="cf" 
+		using-nio="true"
+		type="server"
+		port="1514" 
+		deserializer="rfc6587" />
+
+	<bean id="rfc5424" class="org.springframework.integration.syslog.RFC5424MessageConverter" />
+
+	<bean id="rfc6587" class="org.springframework.integration.syslog.inbound.RFC6587SyslogDeserializer" />
+
 </beans>

spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2014 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.
@@ -37,9 +37,9 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
-import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
 import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
 import org.springframework.integration.syslog.MessageConverter;
+import org.springframework.integration.syslog.RFC5424MessageConverter;
 import org.springframework.integration.syslog.inbound.TcpSyslogReceivingChannelAdapter;
 import org.springframework.integration.syslog.inbound.UdpSyslogReceivingChannelAdapter;
 import org.springframework.integration.test.util.TestUtils;
@@ -81,6 +81,9 @@ public class SyslogReceivingChannelAdapterParserTests {
 	@Autowired
 	private PassThruConverter converter;
 
+	@Autowired
+	private RFC5424MessageConverter rfc5424;
+
 	@Autowired @Qualifier("bar.adapter")
 	private TcpSyslogReceivingChannelAdapter adapter2;
 
@@ -152,7 +155,7 @@ public class SyslogReceivingChannelAdapterParserTests {
 		assertFalse(fullBoatTcp.isAutoStartup());
 		assertEquals(123, fullBoatTcp.getPhase());
 		assertEquals(456L, TestUtils.getPropertyValue(fullBoatUdp, "messagingTemplate.sendTimeout"));
-		assertSame(converter, TestUtils.getPropertyValue(fullBoatTcp, "converter"));
+		assertSame(rfc5424, TestUtils.getPropertyValue(fullBoatTcp, "converter"));
 		assertSame(errors, TestUtils.getPropertyValue(fullBoatTcp, "errorChannel"));
 		assertSame(cf, TestUtils.getPropertyValue(fullBoatTcp, "connectionFactory"));
 	}

spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogDeserializerTests.java

@@ -0,0 +1,141 @@
+/*
+ * Copyright 2014 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.syslog.inbound;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.ByteArrayInputStream;
+import java.util.List;
+import java.util.Map;
+
+import org.junit.Test;
+
+import org.springframework.integration.syslog.SyslogHeaders;
+
+/**
+ * @author Duncan McIntyre
+ * @author Gary Russell
+ * @since 4.1.1
+ *
+ */
+public class SyslogDeserializerTests {
+
+	static final String VALID_UNFRAMED_ENTRY =
+			"<14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - - Removing instance\n";
+	static final String VALID_FRAMED_ENTRY =
+			"106 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - - Removing instance";
+	static final String SHORT_FRAMED_ENTRY =
+			"107 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - - Removing instance";
+	static final String SD_ENTRY_1 =
+			"179 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " +
+					"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] Removing instance";
+	static final String SD_ENTRY_2 =
+			"253 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " +
+					"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"][exampleSDID@32473 " +
+					"iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] Removing instance";
+	static final String SD_ENTRY_3 = "275 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " +
+			"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"][exampleSDID@32473 " +
+			"iut=\\\"3\\\" eventSource=\\\"Application\\\" escapedBracket=\\\"\\]\\\" eventID=\\\"1011\\\"] Removing instance";
+
+	@Test
+	public void shouldParseAValidFramedEntry() throws Exception {
+
+		RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer();
+
+		Map<String, ?> map = deserializer.deserialize(new ByteArrayInputStream(VALID_FRAMED_ENTRY.getBytes()));
+
+		assertEquals(1, map.get(SyslogHeaders.FACILITY));
+		assertEquals(6, map.get(SyslogHeaders.SEVERITY));
+		assertEquals(1, map.get(SyslogHeaders.VERSION));
+		assertEquals("2014-06-20T09:14:07+00:00", map.get(SyslogHeaders.TIMESTAMP));
+		assertEquals("loggregator", map.get(SyslogHeaders.HOST));
+		assertEquals("d0602076-b14a-4c55-852a-981e7afeed38", map.get(SyslogHeaders.APP_NAME));
+		assertEquals("DEA", map.get(SyslogHeaders.PROCID));
+		assertEquals("-", map.get(SyslogHeaders.MSGID));
+		assertEquals("Removing instance", map.get(SyslogHeaders.MESSAGE));
+	}
+
+	@Test
+	public void shouldParseAValidUnframedEntry() throws Exception {
+
+		RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer();
+
+		Map<String, ?> map = deserializer.deserialize(new ByteArrayInputStream(VALID_UNFRAMED_ENTRY.getBytes()));
+
+		assertEquals(1, map.get(SyslogHeaders.FACILITY));
+		assertEquals(6, map.get(SyslogHeaders.SEVERITY));
+		assertEquals(1, map.get(SyslogHeaders.VERSION));
+		assertEquals("2014-06-20T09:14:07+00:00", map.get(SyslogHeaders.TIMESTAMP));
+		assertEquals("loggregator", map.get(SyslogHeaders.HOST));
+		assertEquals("d0602076-b14a-4c55-852a-981e7afeed38", map.get(SyslogHeaders.APP_NAME));
+		assertEquals("DEA", map.get(SyslogHeaders.PROCID));
+		assertEquals("-", map.get(SyslogHeaders.MSGID));
+		assertEquals("Removing instance", map.get(SyslogHeaders.MESSAGE));
+	}
+
+	@Test
+	public void shouldGetStructuredData() throws Exception {
+
+		RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer();
+
+		Map<String, ?> map = deserializer.deserialize(new ByteArrayInputStream(SD_ENTRY_1.getBytes()));
+
+		assertEquals(1, map.get(SyslogHeaders.FACILITY));
+		assertEquals(6, map.get(SyslogHeaders.SEVERITY));
+		assertEquals(1, map.get(SyslogHeaders.VERSION));
+		assertEquals("2014-06-20T09:14:07+00:00", map.get(SyslogHeaders.TIMESTAMP));
+		assertEquals("loggregator", map.get(SyslogHeaders.HOST));
+		assertEquals("d0602076-b14a-4c55-852a-981e7afeed38", map.get(SyslogHeaders.APP_NAME));
+		assertEquals("DEA", map.get(SyslogHeaders.PROCID));
+		assertEquals("-", map.get(SyslogHeaders.MSGID));
+		assertEquals("Removing instance", map.get(SyslogHeaders.MESSAGE));
+		assertEquals(1, ((List<?>) map.get(SyslogHeaders.STRUCTURED_DATA)).size());
+	}
+
+	@Test
+	public void shouldGetMultipleStructuredData() throws Exception {
+
+		RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer();
+
+		Map<String, ?> map = deserializer.deserialize(new ByteArrayInputStream(SD_ENTRY_2.getBytes()));
+
+		assertEquals("false", map.get(SyslogHeaders.DECODE_ERRORS));
+		assertEquals("Removing instance", map.get(SyslogHeaders.MESSAGE));
+		assertEquals(2, ((List<?>) map.get(SyslogHeaders.STRUCTURED_DATA)).size());
+	}
+
+	@Test
+	public void shouldGetMultipleStructuredDataWithEscapedBracket() throws Exception {
+
+		RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer();
+
+		Map<String, ?> map = deserializer.deserialize(new ByteArrayInputStream(SD_ENTRY_3.getBytes()));
+
+		assertEquals("false", map.get(SyslogHeaders.DECODE_ERRORS));
+		assertEquals("Removing instance", map.get(SyslogHeaders.MESSAGE));
+	}
+
+	@Test
+	public void shouldErrorOnShortFramedData() throws Exception {
+
+		RFC6587SyslogDeserializer deserializer = new RFC6587SyslogDeserializer();
+
+		Map<String, ?> map = deserializer.deserialize(new ByteArrayInputStream(SHORT_FRAMED_ENTRY.getBytes()));
+
+		assertEquals("true", map.get(SyslogHeaders.DECODE_ERRORS));
+	}
+
+}

spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogReceivingChannelAdapterTests.java

@@ -29,6 +29,7 @@ import java.net.DatagramPacket;
 import java.net.DatagramSocket;
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -44,7 +45,10 @@ import org.springframework.beans.factory.BeanFactory;
 import org.springframework.context.ApplicationEvent;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
 import org.springframework.integration.syslog.DefaultMessageConverter;
+import org.springframework.integration.syslog.RFC5424MessageConverter;
 import org.springframework.integration.syslog.config.SyslogReceivingChannelAdapterFactoryBean;
 import org.springframework.integration.test.util.SocketUtils;
 import org.springframework.integration.test.util.TestUtils;
@@ -109,7 +113,7 @@ public class SyslogReceivingChannelAdapterTests {
 		Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class));
 		doReturn(true).when(logger).isDebugEnabled();
 		final CountDownLatch sawLog = new CountDownLatch(1);
-		doAnswer(new Answer<Void>(){
+		doAnswer(new Answer<Void>() {
 
 			@Override
 			public Void answer(InvocationOnMock invocation) throws Throwable {
@@ -163,4 +167,91 @@ public class SyslogReceivingChannelAdapterTests {
 		adapter.stop();
 	}
 
+	@Test
+	public void testTcpRFC5424() throws Exception {
+		SyslogReceivingChannelAdapterFactoryBean factory = new SyslogReceivingChannelAdapterFactoryBean(
+				SyslogReceivingChannelAdapterFactoryBean.Protocol.tcp);
+		int port = SocketUtils.findAvailableServerSocket(1514);
+		PollableChannel outputChannel = new QueueChannel();
+		factory.setOutputChannel(outputChannel);
+		ApplicationEventPublisher publisher = mock(ApplicationEventPublisher.class);
+		final CountDownLatch latch = new CountDownLatch(2);
+		doAnswer(new Answer<Object>() {
+
+			@Override
+			public Object answer(InvocationOnMock invocation) throws Throwable {
+				latch.countDown();
+				return null;
+			}
+		}).when(publisher).publishEvent(any(ApplicationEvent.class));
+		factory.setBeanFactory(mock(BeanFactory.class));
+		AbstractServerConnectionFactory connectionFactory = new TcpNioServerConnectionFactory(port);
+		connectionFactory.setDeserializer(new RFC6587SyslogDeserializer());
+		connectionFactory.setApplicationEventPublisher(publisher);
+		factory.setConnectionFactory(connectionFactory);
+		factory.setConverter(new RFC5424MessageConverter());
+		factory.afterPropertiesSet();
+		factory.start();
+		TcpSyslogReceivingChannelAdapter adapter = (TcpSyslogReceivingChannelAdapter) factory.getObject();
+		Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class));
+		doReturn(true).when(logger).isDebugEnabled();
+		final CountDownLatch sawLog = new CountDownLatch(1);
+		doAnswer(new Answer<Void>(){
+
+			@Override
+			public Void answer(InvocationOnMock invocation) throws Throwable {
+				if (((String) invocation.getArguments()[0]).contains("Error on syslog socket")) {
+					sawLog.countDown();
+				}
+				invocation.callRealMethod();
+				return null;
+			}
+		}).when(logger).debug(anyString());
+		new DirectFieldAccessor(adapter).setPropertyValue("logger", logger);
+		Thread.sleep(1000);
+		byte[] buf = ("253 <14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " +
+				"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"]" +
+				"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] Removing instance")
+				.getBytes("UTF-8");
+		Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
+		socket.getOutputStream().write(buf);
+		socket.close();
+		assertTrue(sawLog.await(10, TimeUnit.SECONDS));
+		@SuppressWarnings("unchecked")
+		Message<Map<String, ?>> message = (Message<Map<String, ?>>) outputChannel.receive(10000);
+		assertNotNull(message);
+		assertEquals("loggregator", message.getPayload().get("syslog_HOST"));
+		adapter.stop();
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+	}
+
+	@Test
+	public void testUdpRFC5424() throws Exception {
+		SyslogReceivingChannelAdapterFactoryBean factory = new SyslogReceivingChannelAdapterFactoryBean(
+				SyslogReceivingChannelAdapterFactoryBean.Protocol.udp);
+		int port = SocketUtils.findAvailableUdpSocket(1514);
+		factory.setPort(port);
+		PollableChannel outputChannel = new QueueChannel();
+		factory.setOutputChannel(outputChannel);
+		factory.setBeanFactory(mock(BeanFactory.class));
+		factory.setConverter(new RFC5424MessageConverter());
+		factory.afterPropertiesSet();
+		factory.start();
+		UdpSyslogReceivingChannelAdapter adapter = (UdpSyslogReceivingChannelAdapter) factory.getObject();
+		Thread.sleep(1000);
+		byte[] buf = ("<14>1 2014-06-20T09:14:07+00:00 loggregator d0602076-b14a-4c55-852a-981e7afeed38 DEA - " +
+				"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"]" +
+				"[exampleSDID@32473 iut=\\\"3\\\" eventSource=\\\"Application\\\" eventID=\\\"1011\\\"] Removing instance")
+				.getBytes("UTF-8");
+		DatagramPacket packet = new DatagramPacket(buf, buf.length, new InetSocketAddress("localhost", port));
+		DatagramSocket socket = new DatagramSocket();
+		socket.send(packet);
+		socket.close();
+		@SuppressWarnings("unchecked")
+		Message<Map<String, ?>> message = (Message<Map<String, ?>>) outputChannel.receive(10000);
+		assertNotNull(message);
+		assertEquals("loggregator", message.getPayload().get("syslog_HOST"));
+		adapter.stop();
+	}
+
 }

src/reference/docbook/syslog.xml

@@ -16,17 +16,26 @@
 			Spring Integration 3.0 introduced convenient namespace support for configuring a
 			Syslog inbound adapter in a single element.
 		</para>
+		<para>
+			Starting with <emphasis>version 4.1.1</emphasis>, the framework now supports the
+			extended Syslog format, as specified in <ulink url="https://tools.ietf.org/html/rfc5424"
+			>RFC 5424></ulink>. In addition, when using TCP and RFC5424, both <code>octet counting</code> and
+			<code>non-transparent framing</code> described in <ulink url="https://tools.ietf.org/html/rfc6587"
+			>RFC 6587</ulink> are supported.
+		</para>
 	</section>
 
 	<section id="syslog-inbound-adapter">
 		<title>Syslog &lt;inbound-channel-adapter&gt;</title>
 		<para>
-			This element encompases a <code>UDP</code> or <code>TCP</code> inbound channel adapter
+			This element encompasses a <code>UDP</code> or <code>TCP</code> inbound channel adapter
 			and a <interfacename>MessageConverter</interfacename> to convert the Syslog message to
 			a Spring Integration message. The <classname>DefaultMessageConverter</classname> delegates
 			to the <classname>SyslogToMapTransformer</classname>, creating a message with its payload
 			being the <code>Map</code> of Syslog fields. In addition, all fields except the message
 			are also made available as headers in the message, prefixed with <code>syslog_</code>.
+			In this mode, only <ulink url="https://tools.ietf.org/html/rfc3164">RFC 3164</ulink>
+			(BSD) syslogs are supported.
 		</para>
 		<para>
 			Since <emphasis>version 4.1</emphasis>,
@@ -35,6 +44,25 @@
 			the message payload as the original complete syslog message, in a <code>byte[]</code>,
 			while still setting the headers.
 		</para>
+		<para>
+			Since <emphasis>version 4.1.1</emphasis>, RFC 5424 is also supported, using the
+			<classname>RFC5424MessageConverter</classname>; in this case the fields are not
+			copied as headers, unless <code>asMap</code> is set to <code>false</code>, in which
+			case the original message is the payload and the decoded fields are headers.
+		</para>
+		<important>
+			To use RFC 5424 with a TCP transport, additional configuration is required, to enable
+			the different framing techniques described in RFC 6587. The adapter needs a
+			TCP connection factory configured with a <classname>RFC6587SyslogDeserializer</classname>.
+			By default, this deserializer will handle <code>octet counting</code> and
+			<code>non-transparent framing</code>, using a linefeed (LF) to delimit syslog messages;
+			it uses a <classname>ByteArrayLfSerializer</classname> when <code>octet counting</code>
+			is not detected. To use different <code>non-transparent</code> framing, you can provide
+			it with some other deserializer. While the deserializer can support both <code>octet counting</code>
+			and <code>non-transparent framing</code>, only one form of the latter is supported.
+			If <code>asMap</code> is <code>false</code> on the converter, you must set the
+			<code>retainOriginal</code> constructor argument in the <classname>RFC6587SyslogDeserializer</classname>.
+		</important>
 		<section id="syslog-inbound-examplers">
 			<title>Example Configuration</title>
 			<programlisting language="xml"><![CDATA[<int-syslog:inbound-channel-adapter id="syslogIn" port="1514" />]]></programlisting>
@@ -94,6 +122,27 @@
 				The externally configured <code>connection-factory</code> must be of type <code>server</code> and,
 				the port is defined there rather than on the <code>inbound-channel-adapter</code> element itself.
 			</note>
+			<programlisting language="xml"><![CDATA[<int-syslog:inbound-channel-adapter id="rfc5424Tcp"
+	protocol="tcp"
+	channel="fromSyslog"
+	connection-factory="cf"
+	converter="rfc5424" />
+
+<int-ip:tcp-connection-factory id="cf"
+	using-nio="true"
+	type="server"
+	port="1514"
+	deserializer="rfc6587" />
+
+<bean id="rfc5424" class="org.springframework.integration.syslog.RFC5424MessageConverter" />
+
+<bean id="rfc6587" class="org.springframework.integration.syslog.inbound.RFC6587SyslogDeserializer" />]]></programlisting>
+			<para>
+				A <code>TCP</code> adapter that sends messages to channel <code>fromSyslog</code>. It is configured
+				to use the <code>RFC 5424</code> converter and is configured
+				with a reference to an externally defined connection factory with the <code>RFC 6587</code>
+				deserializer (required for RFC 5424).
+			</para>
 		</section>
 	</section>
 </chapter>