diff --git a/pom.xml b/pom.xml
index 3ac33f3387..5e53acae89 100644
--- a/pom.xml
+++ b/pom.xml
@@ -71,6 +71,15 @@
+
+
+ org.jboss.marshalling
+ jboss-marshalling
+ ${jboss.marshalling.version}
+ compile
+ true
+
+
com.google.protobuf
@@ -198,10 +207,24 @@
1.6.4test
+
+
+ org.jboss.marshalling
+ jboss-marshalling-serial
+ ${jboss.marshalling.version}
+ test
+
+
+ org.jboss.marshalling
+ jboss-marshalling-river
+ ${jboss.marshalling.version}
+ test
+ false
+ 1.3.14.GA
diff --git a/src/main/java/org/jboss/netty/handler/codec/marshalling/ChannelBufferByteInput.java b/src/main/java/org/jboss/netty/handler/codec/marshalling/ChannelBufferByteInput.java
new file mode 100644
index 0000000000..02cc05fabf
--- /dev/null
+++ b/src/main/java/org/jboss/netty/handler/codec/marshalling/ChannelBufferByteInput.java
@@ -0,0 +1,75 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.jboss.netty.handler.codec.marshalling;
+
+import java.io.IOException;
+
+import org.jboss.marshalling.ByteInput;
+import org.jboss.netty.buffer.ChannelBuffer;
+
+/**
+ * {@link ByteInput} implementation which reads its data from a {@link ChannelBuffer}
+ *
+ *
+ */
+public class ChannelBufferByteInput implements ByteInput {
+
+ private final ChannelBuffer buffer;
+
+ public ChannelBufferByteInput(ChannelBuffer buffer) {
+ this.buffer = buffer;
+ }
+
+ public void close() throws IOException {
+ // nothing to do
+ }
+
+ public int available() throws IOException {
+ return buffer.readableBytes();
+ }
+
+ public int read() throws IOException {
+ if (buffer.readable()) {
+ return buffer.readByte() & 0xff;
+ }
+ return -1;
+ }
+
+ public int read(byte[] array) throws IOException {
+ return read(array, 0, array.length);
+ }
+
+ public int read(byte[] dst, int dstIndex, int length) throws IOException {
+ int available = available();
+ if (available == 0) {
+ return -1;
+ }
+
+ length = Math.min(available, length);
+ buffer.readBytes(dst, dstIndex, length);
+ return length;
+ }
+
+ public long skip(long bytes) throws IOException {
+ int readable = buffer.readableBytes();
+ if (readable < bytes) {
+ bytes = readable;
+ }
+ buffer.readerIndex((int) (buffer.readerIndex() + bytes));
+ return bytes;
+ }
+
+}
diff --git a/src/main/java/org/jboss/netty/handler/codec/marshalling/ChannelBufferByteOutput.java b/src/main/java/org/jboss/netty/handler/codec/marshalling/ChannelBufferByteOutput.java
new file mode 100644
index 0000000000..45468a94d3
--- /dev/null
+++ b/src/main/java/org/jboss/netty/handler/codec/marshalling/ChannelBufferByteOutput.java
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.jboss.netty.handler.codec.marshalling;
+
+import java.io.IOException;
+
+import org.jboss.marshalling.ByteOutput;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBufferFactory;
+import org.jboss.netty.buffer.ChannelBuffers;
+
+/**
+ * {@link ByteOutput} implementation which writes the data to a {@link ChannelBuffer}
+ *
+ *
+ */
+public class ChannelBufferByteOutput implements ByteOutput {
+
+ private final ChannelBuffer buffer;
+
+
+ /**
+ * Create a new instance which use the given {@link ChannelBuffer}
+ */
+ public ChannelBufferByteOutput(ChannelBuffer buffer) {
+ this.buffer = buffer;
+ }
+
+ /**
+ * Calls {@link #ChannelBufferByteOutput(ChannelBuffer)} with a dynamic {@link ChannelBuffer}
+ */
+ public ChannelBufferByteOutput(ChannelBufferFactory factory, int estimatedLength) {
+ this(ChannelBuffers.dynamicBuffer(estimatedLength, factory));
+ }
+
+ public void close() throws IOException {
+ // Nothing todo
+ }
+
+ public void flush() throws IOException {
+ // nothing to do
+ }
+
+ public void write(int b) throws IOException {
+ buffer.writeByte(b);
+ }
+
+ public void write(byte[] bytes) throws IOException {
+ buffer.writeBytes(bytes);
+ }
+
+ public void write(byte[] bytes, int srcIndex, int length) throws IOException {
+ buffer.writeBytes(bytes, srcIndex, length);
+ }
+
+ /**
+ * Return the {@link ChannelBuffer} which contains the written content
+ *
+ */
+ public ChannelBuffer getBuffer() {
+ return buffer;
+ }
+
+}
diff --git a/src/main/java/org/jboss/netty/handler/codec/marshalling/LimitingByteInput.java b/src/main/java/org/jboss/netty/handler/codec/marshalling/LimitingByteInput.java
new file mode 100644
index 0000000000..bbb4200b47
--- /dev/null
+++ b/src/main/java/org/jboss/netty/handler/codec/marshalling/LimitingByteInput.java
@@ -0,0 +1,92 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.jboss.netty.handler.codec.marshalling;
+
+import java.io.IOException;
+
+import org.jboss.marshalling.ByteInput;
+
+/**
+ * {@link ByteInput} implementation which wraps another {@link ByteInput} and throws a {@link TooBigObjectException}
+ * if the read limit was reached.
+ *
+ *
+ */
+public class LimitingByteInput implements ByteInput {
+
+ private final ByteInput input;
+ private final long limit;
+ private long read;
+
+ public LimitingByteInput(ByteInput input, long limit) {
+ if (limit <= 0) {
+ throw new IllegalArgumentException("The limit MUST be > 0");
+ }
+ this.input = input;
+ this.limit = limit;
+ }
+
+ public void close() throws IOException {
+ // Nothing todo
+ }
+
+ public int available() throws IOException {
+ int available = input.available();
+ int readable = readable(available);
+ return readable;
+ }
+
+ public int read() throws IOException {
+ int readable = readable(1);
+ if (readable > 0) {
+ int b = input.read();
+ read++;
+ return b;
+ } else {
+ throw new TooBigObjectException();
+ }
+ }
+
+ public int read(byte[] array) throws IOException {
+ return read(array, 0, array.length);
+ }
+
+ public int read(byte[] array, int offset, int length) throws IOException {
+ int readable = readable(length);
+ if (readable > 0) {
+ int i = input.read(array, offset, readable);
+ read += i;
+ return i;
+ } else {
+ throw new TooBigObjectException();
+ }
+ }
+
+ public long skip(long bytes) throws IOException {
+ int readable = readable((int) bytes);
+ if (readable > 0) {
+ long i = input.skip(readable);
+ read += i;
+ return i;
+ } else {
+ throw new TooBigObjectException();
+ }
+ }
+
+ private int readable(int length) {
+ return (int) Math.min(length, limit - read);
+ }
+}
diff --git a/src/main/java/org/jboss/netty/handler/codec/marshalling/MarshallingDecoder.java b/src/main/java/org/jboss/netty/handler/codec/marshalling/MarshallingDecoder.java
new file mode 100644
index 0000000000..2535395a62
--- /dev/null
+++ b/src/main/java/org/jboss/netty/handler/codec/marshalling/MarshallingDecoder.java
@@ -0,0 +1,116 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.jboss.netty.handler.codec.marshalling;
+
+import java.io.IOException;
+import java.io.ObjectStreamConstants;
+
+import org.jboss.marshalling.ByteInput;
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.MarshallingConfiguration;
+import org.jboss.marshalling.Unmarshaller;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ExceptionEvent;
+import org.jboss.netty.handler.codec.replay.ReplayingDecoder;
+import org.jboss.netty.handler.codec.replay.VoidEnum;
+
+/**
+ * {@link ReplayingDecoder} which use an {@link Unmarshaller} to read the Object out of the {@link ChannelBuffer}.
+ *
+ * Most times you want to use {@link ThreadLocalMarshallingDecoder} to get a better performance and less overhead.
+ *
+ *
+ */
+public class MarshallingDecoder extends ReplayingDecoder {
+ protected final MarshallingConfiguration config;
+ protected final MarshallerFactory factory;
+ protected final long maxObjectSize;
+
+ /**
+ * Create a new instance of {@link MarshallingDecoder}.
+ *
+ * @param factory the {@link MarshallerFactory} which is used to obtain the {@link Unmarshaller} from
+ * @param config the {@link MarshallingConfiguration} to use
+ * @param maxObjectSize the maximal size (in bytes) of the {@link Object} to unmarshal. Once the size is exceeded
+ * the {@link Channel} will get closed. Use a a maxObjectSize of <= 0 to disable this.
+ * You should only do this if you are sure that the received Objects will never be big and the
+ * sending side are trusted, as this opens the possibility for a DOS-Attack due an {@link OutOfMemoryError}.
+ *
+ */
+ public MarshallingDecoder(MarshallerFactory factory, MarshallingConfiguration config, long maxObjectSize) {
+ this.factory = factory;
+ this.config = config;
+ this.maxObjectSize = maxObjectSize;
+ }
+
+ @Override
+ protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer, VoidEnum state) throws Exception {
+ Unmarshaller unmarshaller = factory.createUnmarshaller(config);
+ ByteInput input = new ChannelBufferByteInput(buffer);
+ if (maxObjectSize > 0) {
+ input = new LimitingByteInput(input, maxObjectSize);
+ }
+ try {
+ unmarshaller.start(input);
+ Object obj = unmarshaller.readObject();
+ unmarshaller.finish();
+ return obj;
+ } finally {
+ // Call close in a finally block as the ReplayingDecoder will throw an Error if not enough bytes are
+ // readable. This helps to be sure that we do not leak resource
+ unmarshaller.close();
+ }
+ }
+
+ @Override
+ protected Object decodeLast(ChannelHandlerContext ctx, Channel channel,
+ ChannelBuffer buffer, VoidEnum state)
+ throws Exception {
+ switch (buffer.readableBytes()) {
+ case 0:
+ return null;
+ case 1:
+ // Ignore the last TC_RESET
+ if (buffer.getByte(buffer.readerIndex()) == ObjectStreamConstants.TC_RESET) {
+ buffer.skipBytes(1);
+ return null;
+ }
+ }
+
+ Object decoded = decode(ctx, channel, buffer, state);
+ return decoded;
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
+ if (e.getCause() instanceof TooBigObjectException) {
+ e.getChannel().close();
+
+ } else {
+ super.exceptionCaught(ctx, e);
+ }
+ }
+
+ /**
+ * Create a new {@link Unmarshaller} for the given {@link Channel}
+ *
+ */
+ protected Unmarshaller getUnmarshaller(Channel channel) throws IOException {
+ return factory.createUnmarshaller(config);
+ }
+}
diff --git a/src/main/java/org/jboss/netty/handler/codec/marshalling/MarshallingEncoder.java b/src/main/java/org/jboss/netty/handler/codec/marshalling/MarshallingEncoder.java
new file mode 100644
index 0000000000..34d5860524
--- /dev/null
+++ b/src/main/java/org/jboss/netty/handler/codec/marshalling/MarshallingEncoder.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.jboss.netty.handler.codec.marshalling;
+
+import org.jboss.marshalling.Marshaller;
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.MarshallingConfiguration;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelHandler.Sharable;
+import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;
+
+/**
+ * {@link OneToOneEncoder} implementation which uses JBoss Marshalling to marshal
+ * an Object.
+ *
+ * See JBoss Marshalling website
+ * for more informations
+ *
+ */
+@Sharable
+public class MarshallingEncoder extends OneToOneEncoder {
+
+ private final MarshallerFactory factory;
+ private final MarshallingConfiguration config;
+
+
+ public MarshallingEncoder(MarshallerFactory factory, MarshallingConfiguration config) {
+ this.factory = factory;
+ this.config = config;
+ }
+
+
+ @Override
+ protected Object encode(ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
+ Marshaller marshaller = factory.createMarshaller(config);
+ ChannelBufferByteOutput output = new ChannelBufferByteOutput(ctx.getChannel().getConfig().getBufferFactory(), 256);
+ marshaller.start(output);
+ marshaller.writeObject(msg);
+ marshaller.finish();
+ marshaller.close();
+
+ return output.getBuffer();
+ }
+
+}
diff --git a/src/main/java/org/jboss/netty/handler/codec/marshalling/ThreadLocalMarshallingDecoder.java b/src/main/java/org/jboss/netty/handler/codec/marshalling/ThreadLocalMarshallingDecoder.java
new file mode 100644
index 0000000000..9be24a4743
--- /dev/null
+++ b/src/main/java/org/jboss/netty/handler/codec/marshalling/ThreadLocalMarshallingDecoder.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.jboss.netty.handler.codec.marshalling;
+
+import java.io.IOException;
+
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.MarshallingConfiguration;
+import org.jboss.marshalling.Unmarshaller;
+import org.jboss.netty.channel.Channel;
+
+/**
+ * A subclass of {@link MarshallingDecoder} which use one {@link Unmarshaller} per Thread via a {@link ThreadLocal}.
+ *
+ * For more informations see {@link MarshallingDecoder}.
+ *
+ */
+public class ThreadLocalMarshallingDecoder extends MarshallingDecoder {
+
+ private static final ThreadLocal UNMARSHALLERS = new ThreadLocal();
+
+ /**
+ * See {@link MarshallingDecoder#MarshallingDecoder(MarshallerFactory, MarshallingConfiguration, long)}
+ */
+ public ThreadLocalMarshallingDecoder(MarshallerFactory factory, MarshallingConfiguration config, long maxObjectSize) {
+ super(factory, config, maxObjectSize);
+ }
+
+ @Override
+ protected Unmarshaller getUnmarshaller(Channel channel) throws IOException {
+ Unmarshaller unmarshaller = UNMARSHALLERS.get();
+ if (unmarshaller == null) {
+ unmarshaller = factory.createUnmarshaller(config);
+ UNMARSHALLERS.set(unmarshaller);
+ }
+ return unmarshaller;
+ }
+
+}
diff --git a/src/main/java/org/jboss/netty/handler/codec/marshalling/TooBigObjectException.java b/src/main/java/org/jboss/netty/handler/codec/marshalling/TooBigObjectException.java
new file mode 100644
index 0000000000..cc96ab5fe1
--- /dev/null
+++ b/src/main/java/org/jboss/netty/handler/codec/marshalling/TooBigObjectException.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.jboss.netty.handler.codec.marshalling;
+
+import java.io.IOException;
+
+/**
+ * Will get thrown if you try to unmarshal a Object that is to big
+ *
+ */
+public class TooBigObjectException extends IOException {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 533756617517705345L;
+
+ public TooBigObjectException() {
+ super();
+ }
+
+ public TooBigObjectException(String message) {
+ super(message);
+ }
+
+}
diff --git a/src/main/java/org/jboss/netty/handler/codec/marshalling/package-info.java b/src/main/java/org/jboss/netty/handler/codec/marshalling/package-info.java
new file mode 100644
index 0000000000..9e596120c6
--- /dev/null
+++ b/src/main/java/org/jboss/netty/handler/codec/marshalling/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.
+ */
+
+/**
+ * Decoder and Encoder which uses JBoss Marshalling.
+ *
+ */
+package org.jboss.netty.handler.codec.marshalling;
diff --git a/src/test/java/org/jboss/netty/handler/codec/marshalling/AbstractMarshallingDecoderTest.java b/src/test/java/org/jboss/netty/handler/codec/marshalling/AbstractMarshallingDecoderTest.java
new file mode 100644
index 0000000000..bb558faf34
--- /dev/null
+++ b/src/test/java/org/jboss/netty/handler/codec/marshalling/AbstractMarshallingDecoderTest.java
@@ -0,0 +1,134 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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.jboss.netty.handler.codec.marshalling;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+import junit.framework.Assert;
+
+import org.jboss.marshalling.Marshaller;
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.Marshalling;
+import org.jboss.marshalling.MarshallingConfiguration;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.handler.codec.embedder.CodecEmbedderException;
+import org.jboss.netty.handler.codec.embedder.DecoderEmbedder;
+import org.junit.Test;
+
+public abstract class AbstractMarshallingDecoderTest {
+ private final String testObject = new String("test");
+
+ @Test
+ public void testSimpleUnmarshalling() throws IOException {
+ MarshallerFactory marshallerFactory = createMarshallerFactory();
+ MarshallingConfiguration configuration = createMarshallingConfig();
+
+ DecoderEmbedder