Add JBoss Marshalling Encoder/Decoder. See #324

codec/pom.xml

@@ -39,6 +39,23 @@
       <artifactId>protobuf-java</artifactId>
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <groupId>org.jboss.marshalling</groupId>
+      <artifactId>jboss-marshalling</artifactId>
+      <optional>true</optional>
+    </dependency>
+    
+    <!-- Test dependencies for jboss marshalling encoder/decoder -->
+    <dependency>
+      <groupId>org.jboss.marshalling</groupId>
+      <artifactId>jboss-marshalling-serial</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.jboss.marshalling</groupId>
+      <artifactId>jboss-marshalling-river</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 </project>
 

codec/src/main/java/io/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 io.netty.handler.codec.marshalling;
+
+import java.io.IOException;
+
+import org.jboss.marshalling.ByteInput;
+import io.netty.buffer.ChannelBuffer;
+
+/**
+ * {@link ByteInput} implementation which reads its data from a {@link ChannelBuffer}
+ * 
+ *
+ */
+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;
+    }
+
+}

codec/src/main/java/io/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 io.netty.handler.codec.marshalling;
+
+import java.io.IOException;
+
+import org.jboss.marshalling.ByteOutput;
+import io.netty.buffer.ChannelBuffer;
+import io.netty.buffer.ChannelBufferFactory;
+import io.netty.buffer.ChannelBuffers;
+
+/**
+ * {@link ByteOutput} implementation which writes the data to a {@link ChannelBuffer}
+ * 
+ *
+ */
+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;
+    }
+
+}

codec/src/main/java/io/netty/handler/codec/marshalling/LimitingByteInput.java

@@ -0,0 +1,108 @@
+/*
+ * 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 io.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.
+ * 
+ *
+ */
+class LimitingByteInput implements ByteInput {
+    
+    // Use a static instance here to remove the overhead of fillStacktrace
+    private static final TooBigObjectException EXCEPTION = new TooBigObjectException();
+    
+    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 EXCEPTION;
+        }
+    }
+
+    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 EXCEPTION;
+        }
+    }
+
+    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 EXCEPTION;
+        }
+    }
+
+    private int readable(int length) {
+        return (int) Math.min(length, limit - read);
+    }
+    
+    /**
+     * Exception that will get thrown if the {@link Object} is to big to unmarshall
+     *
+     */
+    static final class TooBigObjectException extends IOException {
+
+        /**
+         * 
+         */
+        private static final long serialVersionUID = 1L;
+        
+    }
+}

codec/src/main/java/io/netty/handler/codec/marshalling/MarshallingDecoder.java

@@ -0,0 +1,122 @@
+/*
+ * 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 io.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 io.netty.buffer.ChannelBuffer;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ExceptionEvent;
+import io.netty.handler.codec.frame.TooLongFrameException;
+import io.netty.handler.codec.replay.ReplayingDecoder;
+import io.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<VoidEnum> {
+    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;
+        } catch (LimitingByteInput.TooBigObjectException e) {
+            throw new TooLongFrameException("Object to big to unmarshal");
+        } 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;
+    }
+
+    /**
+     * Calls {@link Channel#close()} if a TooLongFrameException was thrown
+     */
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
+        if (e.getCause() instanceof TooLongFrameException) {
+            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);
+    }
+}

codec/src/main/java/io/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 io.netty.handler.codec.marshalling;
+
+import org.jboss.marshalling.Marshaller;
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.MarshallingConfiguration;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelHandler.Sharable;
+import io.netty.handler.codec.oneone.OneToOneEncoder;
+
+/**
+ * {@link OneToOneEncoder} implementation which uses JBoss Marshalling to marshal 
+ * an Object.
+ * 
+ * See <a href="http://www.jboss.org/jbossmarshalling">JBoss Marshalling website</a> 
+ * 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();
+    }
+
+}

codec/src/main/java/io/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 io.netty.handler.codec.marshalling;
+
+import java.io.IOException;
+
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.MarshallingConfiguration;
+import org.jboss.marshalling.Unmarshaller;
+import io.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<Unmarshaller> UNMARSHALLERS = new ThreadLocal<Unmarshaller>();
+    
+    /**
+     * 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;
+    }
+
+}

codec/src/main/java/io/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 <a href="http://www.jboss.org/jbossmarshalling">JBoss Marshalling</a>.
+ *
+ */
+package io.netty.handler.codec.marshalling;

codec/src/test/java/io/netty/handler/codec/marshalling/AbstractMarshallingDecoderTest.java

@@ -0,0 +1,135 @@
+/*
+ * 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 io.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 io.netty.buffer.ChannelBuffer;
+import io.netty.buffer.ChannelBuffers;
+import io.netty.handler.codec.embedder.CodecEmbedderException;
+import io.netty.handler.codec.embedder.DecoderEmbedder;
+import io.netty.handler.codec.frame.TooLongFrameException;
+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<Object> decoder = new DecoderEmbedder<Object>(createDecoder(marshallerFactory, configuration, 0));
+        
+        ByteArrayOutputStream bout = new ByteArrayOutputStream();
+        Marshaller marshaller = marshallerFactory.createMarshaller(configuration);
+        marshaller.start(Marshalling.createByteOutput(bout));
+        marshaller.writeObject(testObject);
+        marshaller.finish();
+        marshaller.close();
+        
+        byte[] testBytes = bout.toByteArray();
+        
+        decoder.offer(ChannelBuffers.wrappedBuffer(testBytes));
+        assertTrue(decoder.finish());
+        
+        String unmarshalled = (String) decoder.poll();
+        
+        Assert.assertEquals(testObject, unmarshalled);
+        
+        Assert.assertNull(decoder.poll());
+    }
+    
+
+    @Test
+    public void testFragmentedUnmarshalling() throws IOException {
+        MarshallerFactory marshallerFactory = createMarshallerFactory();
+        MarshallingConfiguration configuration = createMarshallingConfig();
+        
+        DecoderEmbedder<Object> decoder = new DecoderEmbedder<Object>(createDecoder(marshallerFactory, configuration, 0));
+        
+        ByteArrayOutputStream bout = new ByteArrayOutputStream();
+        Marshaller marshaller = marshallerFactory.createMarshaller(configuration);
+        marshaller.start(Marshalling.createByteOutput(bout));
+        marshaller.writeObject(testObject);
+        marshaller.finish();
+        marshaller.close();
+        
+        byte[] testBytes = bout.toByteArray();
+        
+        ChannelBuffer buffer = ChannelBuffers.wrappedBuffer(testBytes);
+        ChannelBuffer slice = buffer.readSlice(2);
+        
+        decoder.offer(slice);
+        decoder.offer(buffer);
+        assertTrue(decoder.finish());
+        
+        
+        String unmarshalled = (String) decoder.poll();
+        
+        Assert.assertEquals(testObject, unmarshalled);
+        
+        Assert.assertNull(decoder.poll());
+    }
+    
+    @Test
+    public void testTooBigObject() throws IOException {
+        MarshallerFactory marshallerFactory = createMarshallerFactory();
+        MarshallingConfiguration configuration = createMarshallingConfig();
+        
+        MarshallingDecoder mDecoder = createDecoder(marshallerFactory, configuration, 1);
+        DecoderEmbedder<Object> decoder = new DecoderEmbedder<Object>(mDecoder);
+
+        ByteArrayOutputStream bout = new ByteArrayOutputStream();
+        Marshaller marshaller = marshallerFactory.createMarshaller(configuration);
+        marshaller.start(Marshalling.createByteOutput(bout));
+        marshaller.writeObject(testObject);
+        marshaller.finish();
+        marshaller.close();
+        
+        byte[] testBytes = bout.toByteArray();
+
+        try {
+            decoder.offer(ChannelBuffers.wrappedBuffer(testBytes));
+            fail();
+        } catch (CodecEmbedderException e) {
+            assertEquals(TooLongFrameException.class, e.getCause().getClass());
+            
+        
+        }        
+
+    }
+    
+    protected MarshallingDecoder createDecoder(MarshallerFactory factory, MarshallingConfiguration config, long maxObjectSize) {
+        return new MarshallingDecoder(factory, config, maxObjectSize);
+    }
+    
+    
+    protected abstract MarshallerFactory createMarshallerFactory();
+    protected abstract MarshallingConfiguration createMarshallingConfig();
+
+}

codec/src/test/java/io/netty/handler/codec/marshalling/AbstractMarshallingEncoderTest.java

@@ -0,0 +1,63 @@
+/*
+ * 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 io.netty.handler.codec.marshalling;
+
+import java.io.IOException;
+
+import junit.framework.Assert;
+
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.Marshalling;
+import org.jboss.marshalling.MarshallingConfiguration;
+import org.jboss.marshalling.Unmarshaller;
+import io.netty.buffer.ChannelBuffer;
+import io.netty.handler.codec.embedder.EncoderEmbedder;
+import org.junit.Test;
+
+public abstract class AbstractMarshallingEncoderTest {
+
+    @Test
+    public void testMarshalling() throws IOException, ClassNotFoundException {
+        String testObject = new String("test");
+        
+        final MarshallerFactory marshallerFactory = createMarshallerFactory();
+        final MarshallingConfiguration configuration = createMarshallingConfig();
+        
+        EncoderEmbedder<ChannelBuffer> encoder = new EncoderEmbedder<ChannelBuffer>(new MarshallingEncoder(marshallerFactory, configuration));
+
+        encoder.offer(testObject);
+        Assert.assertTrue(encoder.finish());
+        
+        ChannelBuffer buffer = encoder.poll();
+        
+        Unmarshaller unmarshaller = marshallerFactory.createUnmarshaller(configuration);
+        unmarshaller.start(Marshalling.createByteInput(buffer.toByteBuffer()));
+        String read = (String) unmarshaller.readObject();
+        Assert.assertEquals(testObject, read);
+        
+        Assert.assertEquals(-1, unmarshaller.read());
+
+        Assert.assertNull(encoder.poll());
+        
+        unmarshaller.finish();
+        unmarshaller.close();
+    }
+
+
+    protected abstract MarshallerFactory createMarshallerFactory();
+
+    protected abstract MarshallingConfiguration createMarshallingConfig();
+}

codec/src/test/java/io/netty/handler/codec/marshalling/RiverMarshallingDecoderTest.java

@@ -0,0 +1,37 @@
+/*
+ * 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 io.netty.handler.codec.marshalling;
+
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.Marshalling;
+import org.jboss.marshalling.MarshallingConfiguration;
+
+public class RiverMarshallingDecoderTest extends AbstractMarshallingDecoderTest {
+
+    @Override
+    protected MarshallerFactory createMarshallerFactory() {
+        return Marshalling.getProvidedMarshallerFactory("river");
+    }
+
+    @Override
+    protected MarshallingConfiguration createMarshallingConfig() {
+        // Create a configuration
+        final MarshallingConfiguration configuration = new MarshallingConfiguration();
+        configuration.setVersion(3);
+        return configuration;
+    }
+
+}

codec/src/test/java/io/netty/handler/codec/marshalling/RiverMarshallingEncoderTest.java

@@ -0,0 +1,38 @@
+/*
+ * 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 io.netty.handler.codec.marshalling;
+
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.Marshalling;
+import org.jboss.marshalling.MarshallingConfiguration;
+
+public class RiverMarshallingEncoderTest extends AbstractMarshallingEncoderTest {
+
+
+    @Override
+    protected MarshallerFactory createMarshallerFactory() {
+        return Marshalling.getProvidedMarshallerFactory("river");
+    }
+
+    @Override
+    protected MarshallingConfiguration createMarshallingConfig() {
+        // Create a configuration
+        final MarshallingConfiguration configuration = new MarshallingConfiguration();
+        configuration.setVersion(3);
+        return configuration;
+    }
+
+}

codec/src/test/java/io/netty/handler/codec/marshalling/RiverThreadLocalMarshallingDecoderTest.java

@@ -0,0 +1,27 @@
+/*
+ * 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 io.netty.handler.codec.marshalling;
+
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.MarshallingConfiguration;
+
+public class RiverThreadLocalMarshallingDecoderTest extends RiverMarshallingDecoderTest {
+
+    @Override
+    protected MarshallingDecoder createDecoder(MarshallerFactory factory, MarshallingConfiguration config, long maxObjectSize) {
+        return new ThreadLocalMarshallingDecoder(factory, config, maxObjectSize);
+    }
+}

codec/src/test/java/io/netty/handler/codec/marshalling/SerialMarshallingDecoderTest.java

@@ -0,0 +1,37 @@
+/*
+ * 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 io.netty.handler.codec.marshalling;
+
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.Marshalling;
+import org.jboss.marshalling.MarshallingConfiguration;
+
+public class SerialMarshallingDecoderTest extends AbstractMarshallingDecoderTest {
+
+    @Override
+    protected MarshallerFactory createMarshallerFactory() {
+        return Marshalling.getProvidedMarshallerFactory("serial");
+    }
+
+    @Override
+    protected MarshallingConfiguration createMarshallingConfig() {
+        // Create a configuration
+        final MarshallingConfiguration configuration = new MarshallingConfiguration();
+        configuration.setVersion(5);
+        return configuration;
+    }
+
+}

codec/src/test/java/io/netty/handler/codec/marshalling/SerialMarshallingEncoderTest.java

@@ -0,0 +1,38 @@
+/*
+ * 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 io.netty.handler.codec.marshalling;
+
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.Marshalling;
+import org.jboss.marshalling.MarshallingConfiguration;
+
+public class SerialMarshallingEncoderTest extends AbstractMarshallingEncoderTest {
+
+
+    @Override
+    protected MarshallerFactory createMarshallerFactory() {
+        return Marshalling.getProvidedMarshallerFactory("serial");
+    }
+
+    @Override
+    protected MarshallingConfiguration createMarshallingConfig() {
+        // Create a configuration
+        final MarshallingConfiguration configuration = new MarshallingConfiguration();
+        configuration.setVersion(5);
+        return configuration;
+    }
+
+}

codec/src/test/java/io/netty/handler/codec/marshalling/SerialThreadLocalMarshallingDecoderTest.java

@@ -0,0 +1,28 @@
+/*
+ * 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 io.netty.handler.codec.marshalling;
+
+import org.jboss.marshalling.MarshallerFactory;
+import org.jboss.marshalling.MarshallingConfiguration;
+
+public class SerialThreadLocalMarshallingDecoderTest extends SerialMarshallingDecoderTest {
+
+    @Override
+    protected MarshallingDecoder createDecoder(MarshallerFactory factory, MarshallingConfiguration config, long maxObjectSize) {
+        return new ThreadLocalMarshallingDecoder(factory, config, maxObjectSize);
+    }
+
+}

pom.xml

@@ -83,6 +83,15 @@
 
   <dependencyManagement>
     <dependencies>
+      <!-- JBoss Marshalling - completely optional -->
+      <dependency>
+        <groupId>org.jboss.marshalling</groupId>
+        <artifactId>jboss-marshalling</artifactId>
+        <version>${jboss.marshalling.version}</version>
+        <scope>compile</scope>
+        <optional>true</optional>
+      </dependency>
+    
       <dependency>
         <groupId>com.google.protobuf</groupId>
         <artifactId>protobuf-java</artifactId>
@@ -161,6 +170,19 @@
         </exclusions>
         <optional>true</optional>
       </dependency>
+      <!-- Test dependencies for jboss marshalling encoder/decoder -->
+      <dependency>
+        <groupId>org.jboss.marshalling</groupId>
+        <artifactId>jboss-marshalling-serial</artifactId>
+        <version>${jboss.marshalling.version}</version>
+        <scope>test</scope>
+       </dependency>
+      <dependency>
+        <groupId>org.jboss.marshalling</groupId>
+        <artifactId>jboss-marshalling-river</artifactId>
+        <version>${jboss.marshalling.version}</version>
+        <scope>test</scope>
+      </dependency>
     </dependencies>
   </dependencyManagement>
 
@@ -196,8 +218,13 @@
       <version>1.6.4</version>
       <scope>test</scope>
     </dependency>
+    
   </dependencies>
 
+  <properties>
+    <jboss.marshalling.version>1.3.14.GA</jboss.marshalling.version>
+  </properties>
+  
   <build>
     <plugins>
       <plugin>