public static Pipe newPipe(final MessageUnpacker unpacker, boolean numeric) throws IOException
    {
        MsgpackParser parser = new MsgpackParser(unpacker, numeric);
        final MsgpackInput msgpackInput = new MsgpackInput(parser);
        return new Pipe()
        {
            @Override
            protected Input begin(Pipe.Schema<?> pipeSchema) throws IOException
            {
                return msgpackInput;
            }

            @Override
            protected void end(Pipe.Schema<?> pipeSchema, Input input,
                    boolean cleanupOnly) throws IOException
            {
                if (cleanupOnly)
                {
                    unpacker.close();
                    return;
                }

                assert input == msgpackInput;

                unpacker.close();

            }
        };
    }
Creates a msgpack pipe from a {@link MessageUnpacker}.
protostuff-msgpack/src/main/java/io/protostuff/MsgpackIOUtil.java