diff --git a/engine/src/jmh/java/org/terasology/benchmark/typehandlerlibrary/THLBenchmark.java b/engine/src/jmh/java/org/terasology/benchmark/typehandlerlibrary/THLBenchmark.java
new file mode 100644
index 00000000000..fd5dd2d232a
--- /dev/null
+++ b/engine/src/jmh/java/org/terasology/benchmark/typehandlerlibrary/THLBenchmark.java
@@ -0,0 +1,190 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.benchmark.typehandlerlibrary;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import com.google.gson.Gson;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.reflections.Reflections;
+import org.terasology.engine.persistence.typeHandling.TypeHandlerLibraryImpl;
+import org.terasology.engine.persistence.typeHandling.gson.GsonPersistedDataReader;
+import org.terasology.engine.persistence.typeHandling.gson.GsonPersistedDataSerializer;
+import org.terasology.engine.persistence.typeHandling.gson.GsonPersistedDataWriter;
+import org.terasology.engine.persistence.typeHandling.protobuf.ProtobufDataReader;
+import org.terasology.engine.persistence.typeHandling.protobuf.ProtobufDataWriter;
+import org.terasology.engine.persistence.typeHandling.protobuf.ProtobufPersistedDataSerializer;
+import org.terasology.persistence.serializers.Serializer;
+import org.terasology.persistence.typeHandling.TypeHandlerLibrary;
+import org.terasology.persistence.typeHandling.bytebuffer.ByteBufferDataReader;
+import org.terasology.persistence.typeHandling.bytebuffer.ByteBufferDataWriter;
+import org.terasology.persistence.typeHandling.bytebuffer.ByteBufferPersistedSerializer;
+import org.terasology.persistence.typeHandling.inMemory.InMemoryPersistedDataSerializer;
+import org.terasology.persistence.typeHandling.inMemory.InMemoryReader;
+import org.terasology.persistence.typeHandling.inMemory.InMemoryWriter;
+import org.terasology.reflection.TypeInfo;
+
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 1)
+@Fork(1)
+@Measurement(iterations = 1)
+public class THLBenchmark {
+
+
+    @Benchmark
+    public Optional<byte[]> ser(SerializerState state, Data data) {
+        return state.serializer.serialize(data.value, data.type);
+    }
+
+    @Benchmark
+    public Optional serde(SerializerState state, Data data) {
+        Optional<byte[]> o = state.serializer.serialize(data.value, data.type);
+        return state.serializer.deserialize(data.type, o.get());
+    }
+
+
+    public enum Serializers {
+        GSON((thl) -> {
+            Gson gson = new Gson();
+            return new Serializer(thl, new GsonPersistedDataSerializer(), new GsonPersistedDataWriter(gson),
+                    new GsonPersistedDataReader(gson));
+        }),
+        IN_MEMORY((thl) -> new Serializer(thl, new InMemoryPersistedDataSerializer(), new InMemoryWriter(),
+                new InMemoryReader())),
+        PROTOBUF((thl) -> new Serializer(thl, new ProtobufPersistedDataSerializer(), new ProtobufDataWriter(),
+                new ProtobufDataReader())),
+        BYTEBUFFER((thl) -> new Serializer(thl, new ByteBufferPersistedSerializer(), new ByteBufferDataWriter(),
+                new ByteBufferDataReader()));
+        private final Function<TypeHandlerLibrary, Serializer> creator;
+
+        Serializers(Function<TypeHandlerLibrary, Serializer> creator) {
+            this.creator = creator;
+        }
+    }
+
+    public enum Datas {
+        BOOLEAN(Boolean.TYPE, true),
+        BYTE(Byte.TYPE, (byte) 1),
+        INT(Integer.TYPE, 1),
+        LONG(Long.TYPE, 1L),
+        FLOAT(Float.TYPE, 1F),
+        DOUBLE(Double.TYPE, 1D),
+        CHAR(Character.TYPE, 'c'),
+        STRING(String.class, "string"),
+        BO_ARRAY(boolean[].class, new boolean[]{true}),
+        BY_ARRAY(byte[].class, new byte[]{(byte) 1}),
+        I_ARRAY(int[].class, new int[]{1}),
+        L_ARRAY(long[].class, new long[]{1L}),
+        F_ARRAY(float[].class, new float[]{1F}),
+        D_ARRAY(double[].class, new double[]{1D}),
+        C_ARRAY(char[].class, new char[]{'c'}),
+        S_ARRAY(String[].class, new String[]{"foo", "bar"}),
+        ENUM(ExampleEnum.class, ExampleEnum.ONE),
+        S_LIST(new TypeInfo<List<String>>() {
+        }, Lists.newArrayList("foo", "bar")),
+        S_SET(new TypeInfo<Set<String>>() {
+        }, Sets.newHashSet("foo", "bar")),
+        E_SET(new TypeInfo<EnumSet<ExampleEnum>>() {
+        }, EnumSet.of(ExampleEnum.ONE, ExampleEnum.THREE));
+
+        private final TypeInfo type;
+        private final Object obj;
+
+        Datas(Class clazz, Object obj) {
+            this.type = TypeInfo.of(clazz);
+            this.obj = obj;
+        }
+
+        Datas(TypeInfo type, Object obj) {
+            this.type = type;
+            this.obj = obj;
+        }
+    }
+
+    enum ExampleEnum {
+        ONE,
+        TWO,
+        THREE
+    }
+
+    @State(Scope.Thread)
+    public static class Data {
+        @Param({"BOOLEAN",
+//                "BYTE",
+                "INT",
+//                "LONG",
+//                "FLOAT",
+//                "DOUBLE",
+//                "CHAR",
+//                "STRING",
+                "BO_ARRAY",
+                "BY_ARRAY",
+                "I_ARRAY",
+//                "L_ARRAY",
+//                "F_ARRAY",
+//                "D_ARRAY",
+//                "C_ARRAY",
+                "S_ARRAY",
+//                "ENUM",
+                "S_LIST",
+//                "S_SET",
+                "E_SET"})
+        private static Datas datas;
+
+        TypeInfo type;
+        Object value;
+
+        @Setup(Level.Iteration)
+        public void setup() {
+            type = datas.type;
+            value = datas.obj;
+        }
+    }
+
+    @State(Scope.Benchmark)
+    public static class THLState {
+        private static Reflections reflections;
+        private static TypeHandlerLibrary typeHandlerLibrary;
+
+        @Setup(Level.Trial)
+        public static void setup() {
+            reflections = new Reflections(THLBenchmark.class.getClassLoader());
+            typeHandlerLibrary = TypeHandlerLibraryImpl.withReflections(reflections);
+        }
+    }
+
+    @State(Scope.Benchmark)
+    public static class SerializerState {
+        @Param({"GSON", "IN_MEMORY", "PROTOBUF", "BYTEBUFFER"})
+        private static Serializers serializers;
+
+        private Serializer serializer;
+
+        @Setup(Level.Trial)
+        public void setup(THLState state) {
+            serializer = serializers.creator.apply(THLState.typeHandlerLibrary);
+        }
+    }
+
+
+}
diff --git a/engine/src/main/java/org/terasology/engine/persistence/typeHandling/gson/GsonPersistedDataReader.java b/engine/src/main/java/org/terasology/engine/persistence/typeHandling/gson/GsonPersistedDataReader.java
index 68e2f3b214c..de2f6b2d0c1 100644
--- a/engine/src/main/java/org/terasology/engine/persistence/typeHandling/gson/GsonPersistedDataReader.java
+++ b/engine/src/main/java/org/terasology/engine/persistence/typeHandling/gson/GsonPersistedDataReader.java
@@ -4,12 +4,13 @@
 package org.terasology.engine.persistence.typeHandling.gson;
 
 import com.google.gson.Gson;
-import com.google.gson.JsonObject;
+import com.google.gson.JsonElement;
 import org.terasology.persistence.serializers.PersistedDataReader;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
+import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 
@@ -30,13 +31,18 @@ public GsonPersistedDataReader(Gson gson) {
 
     @Override
     public GsonPersistedData read(InputStream inputStream) throws IOException {
-        JsonObject jsonObject = gson.fromJson(new InputStreamReader(inputStream), JsonObject.class);
+        JsonElement jsonObject = gson.fromJson(new InputStreamReader(inputStream), JsonElement.class);
         return new GsonPersistedData(jsonObject);
     }
 
     @Override
     public GsonPersistedData read(byte[] byteBuffer) throws IOException {
-        JsonObject jsonObject = gson.fromJson(new String(byteBuffer, charset), JsonObject.class);
+        JsonElement jsonObject = gson.fromJson(new String(byteBuffer, charset), JsonElement.class);
         return new GsonPersistedData(jsonObject);
     }
+
+    @Override
+    public GsonPersistedData read(ByteBuffer byteBuffer) throws IOException {
+        throw new UnsupportedOperationException("Idk how to parse this.");
+    }
 }
diff --git a/engine/src/main/java/org/terasology/engine/persistence/typeHandling/gson/GsonPersistedDataWriter.java b/engine/src/main/java/org/terasology/engine/persistence/typeHandling/gson/GsonPersistedDataWriter.java
index e005e7164a0..aa33251fb02 100644
--- a/engine/src/main/java/org/terasology/engine/persistence/typeHandling/gson/GsonPersistedDataWriter.java
+++ b/engine/src/main/java/org/terasology/engine/persistence/typeHandling/gson/GsonPersistedDataWriter.java
@@ -10,6 +10,7 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.OutputStreamWriter;
+import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 
@@ -40,4 +41,9 @@ public void writeTo(GsonPersistedData data, OutputStream outputStream) throws IO
         }
 
     }
+
+    @Override
+    public void writeTo(GsonPersistedData data, ByteBuffer byteBuffer) throws IOException {
+        byteBuffer.put(writeBytes(data));
+    }
 }
diff --git a/engine/src/main/java/org/terasology/engine/persistence/typeHandling/protobuf/ProtobufDataReader.java b/engine/src/main/java/org/terasology/engine/persistence/typeHandling/protobuf/ProtobufDataReader.java
index 8b50c601272..8ec858162e0 100644
--- a/engine/src/main/java/org/terasology/engine/persistence/typeHandling/protobuf/ProtobufDataReader.java
+++ b/engine/src/main/java/org/terasology/engine/persistence/typeHandling/protobuf/ProtobufDataReader.java
@@ -9,6 +9,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.ByteBuffer;
 
 public class ProtobufDataReader implements PersistedDataReader<ProtobufPersistedData> {
     @Override
@@ -22,4 +23,10 @@ public ProtobufPersistedData read(byte[] byteBuffer) throws InvalidProtocolBuffe
         EntityData.Value value = EntityData.Value.parseFrom(byteBuffer);
         return new ProtobufPersistedData(value);
     }
+
+    @Override
+    public ProtobufPersistedData read(ByteBuffer byteBuffer) throws InvalidProtocolBufferException {
+        EntityData.Value value = EntityData.Value.parseFrom(byteBuffer);
+        return new ProtobufPersistedData(value);
+    }
 }
diff --git a/engine/src/main/java/org/terasology/engine/persistence/typeHandling/protobuf/ProtobufDataWriter.java b/engine/src/main/java/org/terasology/engine/persistence/typeHandling/protobuf/ProtobufDataWriter.java
index fce33737756..9d2e00a8cb7 100644
--- a/engine/src/main/java/org/terasology/engine/persistence/typeHandling/protobuf/ProtobufDataWriter.java
+++ b/engine/src/main/java/org/terasology/engine/persistence/typeHandling/protobuf/ProtobufDataWriter.java
@@ -8,6 +8,7 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.nio.ByteBuffer;
 
 public class ProtobufDataWriter implements PersistedDataWriter<ProtobufPersistedData> {
     @Override
@@ -19,4 +20,9 @@ public byte[] writeBytes(ProtobufPersistedData data) {
     public void writeTo(ProtobufPersistedData data, OutputStream outputStream) throws IOException {
         data.getValue().writeTo(CodedOutputStream.newInstance(outputStream));
     }
+
+    @Override
+    public void writeTo(ProtobufPersistedData data, ByteBuffer byteBuffer) throws IOException {
+        byteBuffer.put(data.getValue().toByteArray());
+    }
 }
diff --git a/subsystems/TypeHandlerLibrary/build.gradle.kts b/subsystems/TypeHandlerLibrary/build.gradle.kts
index 4fc9250f370..2ee43ea2ac8 100644
--- a/subsystems/TypeHandlerLibrary/build.gradle.kts
+++ b/subsystems/TypeHandlerLibrary/build.gradle.kts
@@ -31,6 +31,8 @@ dependencies {
     testRuntimeOnly("org.junit.jupiter:junit-jupiter-engine")
     testImplementation("org.mockito:mockito-inline:3.12.4")
 
+    testImplementation("com.google.guava:guava:31.0-jre")
+
     testImplementation("org.mockito:mockito-junit-jupiter:3.12.4")
 }
 
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/PersistedDataReader.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/PersistedDataReader.java
index 13374d77c71..ea219ca8528 100644
--- a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/PersistedDataReader.java
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/PersistedDataReader.java
@@ -7,6 +7,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.ByteBuffer;
 
 /**
  * Read {@link PersistedData} from files, stream, buffer, etc.
@@ -15,4 +16,6 @@ public interface PersistedDataReader<T extends PersistedData> {
     T read(InputStream inputStream) throws IOException;
 
     T read(byte[] byteBuffer) throws IOException;
+
+    T read(ByteBuffer byteBuffer) throws IOException;
 }
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/PersistedDataWriter.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/PersistedDataWriter.java
index 000b07cafe6..0d099621c31 100644
--- a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/PersistedDataWriter.java
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/PersistedDataWriter.java
@@ -7,6 +7,7 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.nio.ByteBuffer;
 
 /**
  * Write {@link PersistedData} to files, stream, buffer, etc.
@@ -16,4 +17,6 @@ public interface PersistedDataWriter<T extends PersistedData> {
     byte[] writeBytes(T data);
 
     void writeTo(T data, OutputStream outputStream) throws IOException;
+
+    void writeTo(T data, ByteBuffer byteBuffer) throws IOException;
 }
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/Serializer.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/Serializer.java
index e4a0586cc04..caf0385885f 100644
--- a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/Serializer.java
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/serializers/Serializer.java
@@ -12,15 +12,15 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.ByteBuffer;
 import java.util.Optional;
 
 /**
- * The abstract class that all serializers derive from. It by default provides the ability to serialize/deserialize an
- * object to/from a {@link PersistedData} using the given {@link PersistedDataSerializer}.
+ * The abstract class that all serializers derive from. It by default provides the ability to serialize/deserialize an object to/from a
+ * {@link PersistedData} using the given {@link PersistedDataSerializer}.
  * <p>
- * Implementors simply need to specify the type of {@link PersistedDataSerializer} to use and can provide convenience
- * methods that use {@link #serializeToPersisted(Object, TypeInfo)} and {@link #deserializeFromPersisted(PersistedData,
- * TypeInfo)}.
+ * Implementors simply need to specify the type of {@link PersistedDataSerializer} to use and can provide convenience methods that use
+ * {@link #serializeToPersisted(Object, TypeInfo)} and {@link #deserializeFromPersisted(PersistedData, TypeInfo)}.
  */
 public final class Serializer<D extends PersistedData> {
 
@@ -32,7 +32,7 @@ public final class Serializer<D extends PersistedData> {
     private final PersistedDataReader<D> reader;
 
     public Serializer(TypeHandlerLibrary typeHandlerLibrary, PersistedDataSerializer persistedDataSerializer,
-                         PersistedDataWriter<D> writer, PersistedDataReader<D> reader) {
+                      PersistedDataWriter<D> writer, PersistedDataReader<D> reader) {
         this.typeHandlerLibrary = typeHandlerLibrary;
         this.persistedDataSerializer = persistedDataSerializer;
         this.writer = writer;
@@ -40,15 +40,14 @@ public Serializer(TypeHandlerLibrary typeHandlerLibrary, PersistedDataSerializer
     }
 
     /**
-     * Serializes the given object to a {@link PersistedData} using the stored {@link #persistedDataSerializer} by
-     * loading a {@link org.terasology.persistence.typeHandling.TypeHandler TypeHandler} from the {@link
-     * #typeHandlerLibrary}.
+     * Serializes the given object to a {@link PersistedData} using the stored {@link #persistedDataSerializer} by loading a {@link
+     * org.terasology.persistence.typeHandling.TypeHandler TypeHandler} from the {@link #typeHandlerLibrary}.
      *
      * @param object The object to serialize.
      * @param typeInfo A {@link TypeInfo} specifying the type of the object to serialize.
      * @param <T> The type of the object to serialize.
-     * @return A {@link PersistedData}, if the serialization was successful. Serialization usually fails only because an
-     *         appropriate type handler could not be found for the given type.
+     * @return A {@link PersistedData}, if the serialization was successful. Serialization usually fails only because an appropriate type
+     *         handler could not be found for the given type.
      */
     private <T> Optional<D> serializeToPersisted(T object, TypeInfo<T> typeInfo) {
         return typeHandlerLibrary.getTypeHandler(typeInfo)
@@ -56,16 +55,15 @@ private <T> Optional<D> serializeToPersisted(T object, TypeInfo<T> typeInfo) {
     }
 
     /**
-     * Deserializes an object of the given type from a {@link PersistedData} using the stored {@link
-     * #persistedDataSerializer} by loading a {@link org.terasology.persistence.typeHandling.TypeHandler TypeHandler}
-     * from the {@link #typeHandlerLibrary}.
+     * Deserializes an object of the given type from a {@link PersistedData} using the stored {@link #persistedDataSerializer} by loading a
+     * {@link org.terasology.persistence.typeHandling.TypeHandler TypeHandler} from the {@link #typeHandlerLibrary}.
      *
      * @param data The {@link PersistedData} containing the serialized representation of the object.
      * @param typeInfo The {@link TypeInfo} specifying the type to deserialize the object as.
      * @param <T> The type to deserialize the object as.
-     * @return The deserialized object of type {@link T}, if the deserialization was successful. Deserialization usually
-     *         fails when an appropriate type handler could not be found for the type {@link T} <i>or</i> if the
-     *         serialized object representation in {@code data} does not represent an object of type {@link T}.
+     * @return The deserialized object of type {@link T}, if the deserialization was successful. Deserialization usually fails when an
+     *         appropriate type handler could not be found for the type {@link T} <i>or</i> if the serialized object representation in
+     *         {@code data} does not represent an object of type {@link T}.
      */
     private <T> Optional<T> deserializeFromPersisted(D data, TypeInfo<T> typeInfo) {
         return typeHandlerLibrary.getTypeHandler(typeInfo).flatMap(typeHandler -> typeHandler.deserialize(data));
@@ -107,14 +105,32 @@ public <T> Optional<T> deserialize(TypeInfo<T> type, byte[] bytes) {
         return Optional.empty();
     }
 
+    /**
+     * Deserialize an object of type from {@link ByteBuffer}
+     *
+     * @param type The {@link TypeInfo} specifying the type to deserialize the object as.
+     * @param byteBuffer The ByteBuffer contains data for deserialization.
+     * @param <T> The type to deserialize the object as.
+     * @return an Object if deserialization is success, empty otherwise
+     */
+    public <T> Optional<T> deserialize(TypeInfo<T> type, ByteBuffer byteBuffer) {
+        try {
+            D persistedData = reader.read(byteBuffer);
+            return deserializeFromPersisted(persistedData, type);
+        } catch (IOException e) {
+            logger.error("Cannot deserialize type [" + type + "]", e);
+        }
+        return Optional.empty();
+    }
+
     /**
      * Serializes the given object to a ByteArray
      *
      * @param object The object to serialize.
      * @param type A {@link TypeInfo} specifying the type of the object to serialize.
      * @param <T> The type of the object to serialize.
-     * @return A ByteArray, if the serialization was successful. Serialization usually fails only because an appropriate
-     *         type handler could not be found for the given type.
+     * @return A ByteArray, if the serialization was successful. Serialization usually fails only because an appropriate type handler could
+     *         not be found for the given type.
      */
     public <T> Optional<byte[]> serialize(T object, TypeInfo<T> type) {
         Optional<D> persistedData = serializeToPersisted(object, type);
@@ -146,4 +162,30 @@ public <T> void serialize(T object, TypeInfo<T> type, OutputStream outputStream)
             logger.error("Cannot serialize [{}]", type);
         }
     }
+
+    /**
+     * Serializes and write the given object to  {@link ByteBuffer}
+     *
+     * @param object The object to serialize.
+     * @param type A {@link TypeInfo} specifying the type of the object to serialize.
+     * @param byteBuffer A {@link ByteBuffer} which will used for writing.
+     * @param <T> The type of the object to serialize.
+     */
+    public <T> void serialize(T object, TypeInfo<T> type, ByteBuffer byteBuffer) {
+        Optional<D> persistedData = serializeToPersisted(object, type);
+        if (persistedData.isPresent()) {
+            try {
+                writer.writeTo(persistedData.get(), byteBuffer);
+            } catch (IOException e) {
+                logger.error("Cannot serialize [" + type + "]", e);
+            }
+        } else {
+            logger.error("Cannot serialize [{}]", type);
+        }
+    }
+
+    @Override
+    public String toString() {
+        return String.format("Serializer with %s", persistedDataSerializer.getClass().getSimpleName());
+    }
 }
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/BBConsts.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/BBConsts.java
new file mode 100644
index 00000000000..898271551fc
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/BBConsts.java
@@ -0,0 +1,22 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling.bytebuffer;
+
+/**
+ * ByteBuffer serializer constants.
+ */
+public final  class BBConsts {
+    public static final int DOUBLE_SIZE = 8;
+    public static final int BYTE_SIZE = 1;
+    public static final int INT_SIZE = 4;
+    public static final int FLOAT_SIZE = 4;
+    public static final int LONG_SIZE = 8;
+
+    public static final int TYPE_HEADER = 1;
+    public static final int SIZE_HEADER = INT_SIZE;
+    private BBConsts() {
+
+    }
+
+}
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/BBType.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/BBType.java
new file mode 100644
index 00000000000..d6d1fb3a5a6
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/BBType.java
@@ -0,0 +1,80 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling.bytebuffer;
+
+/**
+ * Type codes for array types.
+ */
+public enum BBType {
+    NULL(0),
+    BOOLEAN(1),
+    FLOAT(2),
+    DOUBLE(3),
+    LONG(4),
+    INTEGER(5),
+    STRING(6),
+    BYTES(7),
+    BYTEBUFFER(8),
+    BOOLEAN_ARRAY(9),
+    FLOAT_ARRAY(10),
+    DOUBLE_ARRAY(11),
+    LONG_ARRAY(12),
+    INTEGER_ARRAY(13),
+    STRING_ARRAY(14),
+    VALUE_ARRAY(15),
+    VALUEMAP(16);
+
+    private final byte code;
+
+    BBType(int code) {
+        this.code = (byte) code;
+    }
+
+    public static BBType parse(byte code) {
+        for (BBType type : values()) {
+            if (type.code == code) {
+                return type;
+            }
+        }
+        return null;
+    }
+
+
+    public BBType getPrimitiveType() {
+        switch (this) {
+            case BOOLEAN_ARRAY:
+                return BBType.BOOLEAN;
+            case FLOAT_ARRAY:
+                return BBType.FLOAT;
+            case DOUBLE_ARRAY:
+                return BBType.DOUBLE;
+            case LONG_ARRAY:
+                return BBType.LONG;
+            case INTEGER_ARRAY:
+                return BBType.INTEGER;
+            case STRING_ARRAY:
+                return BBType.STRING;
+        }
+        return null;
+    }
+
+    public boolean isArray() {
+        switch (this) {
+            case BOOLEAN_ARRAY:
+            case FLOAT_ARRAY:
+            case DOUBLE_ARRAY:
+            case LONG_ARRAY:
+            case INTEGER_ARRAY:
+            case STRING_ARRAY:
+            case VALUE_ARRAY:
+                return true;
+            default:
+                return false;
+        }
+    }
+
+    public byte getCode() {
+        return code;
+    }
+}
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferDataReader.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferDataReader.java
new file mode 100644
index 00000000000..cee648c158b
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferDataReader.java
@@ -0,0 +1,27 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling.bytebuffer;
+
+import org.terasology.persistence.serializers.PersistedDataReader;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+
+public class ByteBufferDataReader implements PersistedDataReader<ByteBufferPersistedData> {
+    @Override
+    public ByteBufferPersistedData read(InputStream inputStream) throws IOException {
+        throw new UnsupportedOperationException("Idk which size are you using");
+    }
+
+    @Override
+    public ByteBufferPersistedData read(byte[] byteBuffer) throws IOException {
+        return new ByteBufferPersistedData(ByteBuffer.wrap(byteBuffer).asReadOnlyBuffer());
+    }
+
+    @Override
+    public ByteBufferPersistedData read(ByteBuffer byteBuffer) throws IOException {
+        return new ByteBufferPersistedData(byteBuffer);
+    }
+}
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferDataWriter.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferDataWriter.java
new file mode 100644
index 00000000000..2a77623cc29
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferDataWriter.java
@@ -0,0 +1,27 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling.bytebuffer;
+
+import org.terasology.persistence.serializers.PersistedDataWriter;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+public class ByteBufferDataWriter implements PersistedDataWriter<ByteBufferPersistedData> {
+    @Override
+    public byte[] writeBytes(ByteBufferPersistedData data) {
+        return data.getData().array();
+    }
+
+    @Override
+    public void writeTo(ByteBufferPersistedData data, OutputStream outputStream) throws IOException {
+        throw new UnsupportedOperationException("Idk which size are you using");
+    }
+
+    @Override
+    public void writeTo(ByteBufferPersistedData data, ByteBuffer byteBuffer) throws IOException {
+        byteBuffer.put(data.getAsByteBuffer());
+    }
+}
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedData.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedData.java
new file mode 100644
index 00000000000..3150f204941
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedData.java
@@ -0,0 +1,236 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling.bytebuffer;
+
+import org.terasology.persistence.typeHandling.DeserializationException;
+import org.terasology.persistence.typeHandling.PersistedData;
+import org.terasology.persistence.typeHandling.PersistedDataArray;
+import org.terasology.persistence.typeHandling.PersistedDataMap;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * ByteBuffer-backed persisted data representation.
+ * <pre>
+ * 1 byte - BBType
+ * 0..n bytes - data
+ * </pre>
+ */
+public class ByteBufferPersistedData implements PersistedData {
+    protected final ByteBuffer byteBuffer;
+    protected final int position;
+    protected final BBType type;
+    private final boolean typeForced;
+
+    public ByteBufferPersistedData(ByteBuffer byteBuffer, int position, byte type) {
+        this.byteBuffer = byteBuffer;
+        this.position = position;
+        this.type = BBType.parse(type);
+        this.typeForced = true;
+    }
+
+    public ByteBufferPersistedData(ByteBuffer byteBuffer) {
+        this.byteBuffer = byteBuffer;
+        this.position = byteBuffer.position();
+        this.type = BBType.parse(byteBuffer.get());
+        this.typeForced = false;
+    }
+
+    public ByteBufferPersistedData(ByteBuffer byteBuffer, int position) {
+        this.byteBuffer = byteBuffer;
+        this.position = position;
+        byteBuffer.position(position);
+        this.type = BBType.parse(byteBuffer.get());
+        this.typeForced = false;
+    }
+
+    public ByteBufferPersistedData(ByteBuffer byteBuffer, byte type) {
+        this(byteBuffer, byteBuffer.position(), type);
+    }
+
+
+    public ByteBuffer getData() {
+        return byteBuffer;
+    }
+
+    @Override
+    public String getAsString() {
+        if (!isString()) {
+            throw new ClassCastException(String.format("Source is not of type string: %s", type.name()));
+        }
+        resetPosition();
+        int size = byteBuffer.getInt();
+        byte[] bytes = new byte[size];
+        byteBuffer.get(bytes);
+        return new String(bytes, StandardCharsets.UTF_8);
+    }
+
+    private void resetPosition() {
+        byteBuffer.position(position + (typeForced ? 0 : 1));
+    }
+
+    @Override
+    public double getAsDouble() {
+        if (!isNumber()) {
+            throw new ClassCastException(String.format("Source is not of type number: %s", type.name()));
+        }
+        resetPosition();
+        switch (type) {
+            case INTEGER:
+                return getData().getInt();
+            case FLOAT:
+                return getData().getFloat();
+            case DOUBLE:
+                return getData().getDouble();
+            case LONG:
+                return (double) getData().getLong();
+        }
+        throw new ClassCastException(String.format("Source is not of type number: %s", type.name()));
+    }
+
+    @Override
+    public float getAsFloat() {
+        if (!isNumber()) {
+            throw new ClassCastException(String.format("Source is not of type number: %s", type.name()));
+        }
+        resetPosition();
+        switch (type) {
+            case INTEGER:
+                return (float) getData().getInt();
+            case FLOAT:
+                return getData().getFloat();
+            case DOUBLE:
+                return (float) getData().getDouble();
+            case LONG:
+                return (float) getData().getLong();
+        }
+        throw new ClassCastException(String.format("Source is not of type number: %s", type.name()));
+    }
+
+    @Override
+    public int getAsInteger() {
+        if (!isNumber()) {
+            throw new ClassCastException(String.format("Source is not of type number: %s", type.name()));
+        }
+        resetPosition();
+        switch (type) {
+            case INTEGER:
+                return getData().getInt();
+            case FLOAT:
+                return (int) getData().getFloat();
+            case DOUBLE:
+                return (int) getData().getDouble();
+            case LONG:
+                return (int) getData().getLong();
+        }
+        throw new ClassCastException(String.format("Source is not of type number: %s", type.name()));
+    }
+
+    @Override
+    public long getAsLong() {
+        if (!isNumber()) {
+            throw new ClassCastException(String.format("Source is not of type number: %s", type.name()));
+        }
+        resetPosition();
+        switch (type) {
+            case INTEGER:
+                return getData().getInt();
+            case FLOAT:
+                return (long) getData().getFloat();
+            case DOUBLE:
+                return (long) getData().getDouble();
+            case LONG:
+                return getData().getLong();
+        }
+        throw new ClassCastException(String.format("Source is not of type number: %s", type.name()));
+    }
+
+
+    @Override
+    public boolean getAsBoolean() {
+        if (!isBoolean()) {
+            throw new ClassCastException(String.format("Source is not of type boolean: %s", type.name()));
+        }
+        resetPosition();
+        return byteBuffer.get() != 0; // Don't Packed booleans
+    }
+
+    @Override
+    public byte[] getAsBytes() {
+        if (!isBytes()) {
+            throw new DeserializationException(String.format("Source is not of type bytes or bytebuffer: %s", type.name()));
+        }
+        resetPosition();
+        int size = byteBuffer.getInt();
+        byte[] bytes = new byte[size];
+        byteBuffer.get(bytes);
+        return bytes;
+    }
+
+    @Override
+    public ByteBuffer getAsByteBuffer() {
+        if (!isBytes()) {
+            throw new DeserializationException(String.format("Source is not of type bytes or bytebuffer: %s", type.name()));
+        }
+        resetPosition();
+        return ByteBuffer.wrap(getAsBytes());
+    }
+
+    @Override
+    public PersistedDataArray getAsArray() {
+        if (!isArray()) {
+            throw new IllegalStateException(String.format("Source is not of type array: %s", type.name()));
+        }
+        byteBuffer.position(position);
+        return new ByteBufferPersistedDataArray(byteBuffer);
+    }
+
+    @Override
+    public PersistedDataMap getAsValueMap() {
+        if (!isValueMap()) {
+            throw new IllegalStateException(String.format("Source is not of type valuemap: %s", type.name()));
+        }
+        byteBuffer.position(position);
+        return new ByteBufferPersistedDataMap(byteBuffer);
+    }
+
+    @Override
+    public boolean isString() {
+        return type == BBType.STRING;
+    }
+
+    @Override
+    public boolean isNumber() {
+        return type == BBType.FLOAT
+                || type == BBType.DOUBLE
+                || type == BBType.INTEGER
+                || type == BBType.LONG;
+    }
+
+    @Override
+    public boolean isBoolean() {
+        return type == BBType.BOOLEAN;
+    }
+
+    @Override
+    public boolean isBytes() {
+        return type == BBType.BYTES || type == BBType.BYTEBUFFER;
+    }
+
+    @Override
+    public boolean isArray() {
+        return type.isArray();
+    }
+
+    @Override
+    public boolean isValueMap() {
+        return type == BBType.VALUEMAP;
+    }
+
+    @Override
+    public boolean isNull() {
+        return type == BBType.NULL;
+    }
+}
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedDataArray.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedDataArray.java
new file mode 100644
index 00000000000..e34cac48c61
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedDataArray.java
@@ -0,0 +1,364 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling.bytebuffer;
+
+import gnu.trove.list.TDoubleList;
+import gnu.trove.list.TFloatList;
+import gnu.trove.list.TIntList;
+import gnu.trove.list.TLongList;
+import gnu.trove.list.array.TDoubleArrayList;
+import gnu.trove.list.array.TFloatArrayList;
+import gnu.trove.list.array.TIntArrayList;
+import gnu.trove.list.array.TLongArrayList;
+import org.terasology.persistence.typeHandling.DeserializationException;
+import org.terasology.persistence.typeHandling.PersistedData;
+import org.terasology.persistence.typeHandling.PersistedDataArray;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+/**
+ * ByteBuffer-backed persisted data array representation.
+ * <pre>
+ * 1 byte - BBType
+ * 4 bytes - size
+ * 0..n bytes - data
+ * </pre>
+ */
+public class ByteBufferPersistedDataArray extends ByteBufferPersistedData implements PersistedDataArray {
+
+    private final int size;
+
+    public ByteBufferPersistedDataArray(ByteBuffer byteBuffer) {
+        super(byteBuffer);
+        size = byteBuffer.getInt();
+    }
+
+    @Override
+    public int size() {
+        return size;
+    }
+
+    @Override
+    public PersistedData getArrayItem(int index) {
+        BBType primitiveType = type.getPrimitiveType();
+        if (primitiveType != null) {
+            return new ByteBufferPersistedData(byteBuffer, calculateIndex(index), primitiveType.getCode());
+        } else {
+            return new ByteBufferPersistedData(byteBuffer, calculateIndex(index));
+        }
+    }
+
+    @Override
+    public boolean isNumberArray() {
+        return type == BBType.FLOAT_ARRAY
+                || type == BBType.DOUBLE_ARRAY
+                || type == BBType.INTEGER_ARRAY
+                || type == BBType.LONG_ARRAY;
+    }
+
+    @Override
+    public boolean isBooleanArray() {
+        return type == BBType.BOOLEAN_ARRAY;
+    }
+
+    @Override
+    public boolean isStringArray() {
+        return type == BBType.STRING_ARRAY;
+    }
+
+    @Override
+    public List<String> getAsStringArray() {
+        List<String> list = new ArrayList<>(size());
+        for (int i = 0; i < size(); i++) {
+            list.add(getArrayItem(i).getAsString());
+        }
+        return list;
+    }
+
+    @Override
+    public String getAsString() {
+        if (isStringArray()) {
+            if (size() == 1) {
+                return getArrayItem(0).getAsString();
+            } else {
+                throw new IllegalStateException("String array have size != 1");
+            }
+        } else {
+            if (size() == 1) {
+                PersistedData data = getArrayItem(0);
+                if (data.isString()) {
+                    return data.getAsString();
+                }
+            }
+            throw new ClassCastException(String.format("Source is not of type string array: %s", type.name()));
+        }
+    }
+
+
+    @Override
+    public double getAsDouble() {
+        if (isNumberArray()) {
+            if (size() == 1) {
+                return getArrayItem(0).getAsDouble();
+            } else {
+                throw new IllegalStateException("Number array have size != 1");
+            }
+        } else {
+            if (size() == 1) {
+                PersistedData data = getArrayItem(0);
+                if (data.isNumber()) {
+                    return data.getAsDouble();
+                }
+            }
+            throw new ClassCastException(String.format("Source is not of type number array: %s", type.name()));
+        }
+    }
+
+    @Override
+    public float getAsFloat() {
+        if (isNumberArray()) {
+            if (size() == 1) {
+                return getArrayItem(0).getAsFloat();
+            } else {
+                throw new IllegalStateException("Number array have size != 1");
+            }
+        } else {
+            if (size() == 1) {
+                PersistedData data = getArrayItem(0);
+                if (data.isNumber()) {
+                    return data.getAsFloat();
+                }
+            }
+            throw new ClassCastException(String.format("Source is not of type number array: %s", type.name()));
+        }
+    }
+
+    @Override
+    public int getAsInteger() {
+        if (isNumberArray()) {
+            if (size() == 1) {
+                return getArrayItem(0).getAsInteger();
+            } else {
+                throw new IllegalStateException("Number array have size != 1");
+            }
+        } else {
+            if (size() == 1) {
+                PersistedData data = getArrayItem(0);
+                if (data.isNumber()) {
+                    return data.getAsInteger();
+                }
+            }
+            throw new ClassCastException(String.format("Source is not of type number array: %s", type.name()));
+        }
+    }
+
+    @Override
+    public long getAsLong() {
+        if (isNumberArray()) {
+            if (size() == 1) {
+                return getArrayItem(0).getAsLong();
+            } else {
+                throw new IllegalStateException("Number array have size != 1");
+            }
+        } else {
+            if (size() == 1) {
+                PersistedData data = getArrayItem(0);
+                if (data.isNumber()) {
+                    return data.getAsLong();
+                }
+            }
+            throw new ClassCastException(String.format("Source is not of type number array: %s", type.name()));
+        }
+    }
+
+    @Override
+    public TDoubleList getAsDoubleArray() {
+        byteBuffer.position(position + BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER);
+        TDoubleList list = new TDoubleArrayList(size());
+        Iterator<PersistedData> iter = typedIterator(type.getPrimitiveType());
+        while (iter.hasNext()) {
+            list.add(iter.next().getAsDouble());
+        }
+        return list;
+    }
+
+    @Override
+    public TFloatList getAsFloatArray() {
+        byteBuffer.position(position + BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER);
+        TFloatList list = new TFloatArrayList(size());
+        Iterator<PersistedData> iter = typedIterator(type.getPrimitiveType());
+        while (iter.hasNext()) {
+            list.add(iter.next().getAsFloat());
+        }
+        return list;
+    }
+
+    @Override
+    public TIntList getAsIntegerArray() {
+        byteBuffer.position(position + BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER);
+        TIntList list = new TIntArrayList(size());
+        Iterator<PersistedData> iter = typedIterator(type.getPrimitiveType());
+        while (iter.hasNext()) {
+            list.add(iter.next().getAsInteger());
+        }
+        return list;
+    }
+
+    @Override
+    public TLongList getAsLongArray() {
+        byteBuffer.position(position + BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER);
+        TLongList list = new TLongArrayList(size());
+        Iterator<PersistedData> iter = typedIterator(type.getPrimitiveType());
+        while (iter.hasNext()) {
+            list.add(iter.next().getAsLong());
+        }
+        return list;
+    }
+
+    @Override
+    public boolean[] getAsBooleanArray() {
+        byteBuffer.position(position + BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER);
+        int sizeInBytes = size() % 8 + 1;
+        byte[] bytes = new byte[sizeInBytes];
+        byteBuffer.get(bytes);
+        boolean[] booleans = new boolean[size()];
+        for (int i = 0; i < sizeInBytes; i++) {
+            for (int bi = 0; bi < 8; bi++) {
+                if (i * 8 + bi >= size()) {
+                    break;
+                }
+                booleans[i * 8 + bi] = ((bytes[i] >> bi) & 1) == 1;
+            }
+        }
+        return booleans;
+    }
+
+    @Override
+    public List<PersistedData> getAsValueArray() {
+        byteBuffer.position(position + BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER);
+        List<PersistedData> data = new ArrayList<>(size());
+        for (int i = 0; i < size(); i++) {
+            data.add(new ByteBufferPersistedData(byteBuffer, calculateIndex(i)));
+        }
+        return data;
+    }
+
+    @Override
+    public Iterator<PersistedData> iterator() {
+        return new Iterator<>() {
+            private int index;
+
+            @Override
+            public boolean hasNext() {
+                return index < size;
+            }
+
+            @Override
+            public PersistedData next() {
+                if (!hasNext()) {
+                    throw new NoSuchElementException("iterator haven't something.");
+                }
+                PersistedData data = new ByteBufferPersistedData(byteBuffer, calculateIndex(index));
+                index++;
+                return data;
+            }
+        };
+    }
+
+    private Iterator<PersistedData> typedIterator(BBType type) {
+        return new Iterator<>() {
+            private int index;
+
+            @Override
+            public boolean hasNext() {
+                return index < size;
+            }
+
+            @Override
+            public PersistedData next() {
+                if (!hasNext()) {
+                    throw new NoSuchElementException("iterator haven't something.");
+                }
+                PersistedData data = new ByteBufferPersistedData(byteBuffer, calculateIndex(index), type.getCode());
+                index++;
+                return data;
+            }
+        };
+    }
+
+    private int calculateIndex(int index) {
+        switch (type) {
+            case BOOLEAN_ARRAY:
+                return BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + index % 8 + 1;
+            case FLOAT_ARRAY:
+            case INTEGER_ARRAY:
+                return BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + index * 4;
+            case DOUBLE_ARRAY:
+            case LONG_ARRAY:
+                return BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + index * 8;
+            case STRING_ARRAY: {
+                int pos = position + BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER;
+                for (int i = 0; i < index; i++) {
+                    pos += byteBuffer.getInt(pos) + BBConsts.SIZE_HEADER;
+                }
+                return pos;
+            }
+            case VALUE_ARRAY: {
+                int pos = 0;
+                for (int i = 0; i < index; i++) {
+                    pos += byteBuffer.getInt(position + BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + i *  BBConsts.SIZE_HEADER);
+                }
+                int sizeArraySize = size() * 4;
+                return pos + position + BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + sizeArraySize;
+            }
+
+        }
+        throw new UnsupportedOperationException("IDK how it to do");
+    }
+
+    @Override
+    public boolean getAsBoolean() {
+        if (isBooleanArray()) {
+            if (size() == 1) {
+                return (byteBuffer.get(position + BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER) & 1) == 1;
+            } else {
+                throw new IllegalStateException("boolean array have size != 1");
+            }
+        } else {
+            if (size() == 1) {
+                PersistedData data = getArrayItem(0);
+                if (data.isBoolean()) {
+                    return data.getAsBoolean();
+                }
+            }
+            throw new ClassCastException(String.format("Source is not of type boolean array: %s", type.name()));
+        }
+    }
+
+    @Override
+    public byte[] getAsBytes() {
+        if (size() == 1) {
+            PersistedData data = getArrayItem(0);
+            if (data.isBytes()) {
+                return data.getAsBytes();
+            }
+        }
+        throw new DeserializationException(String.format("Source is not of type bytes array: %s", type.name()));
+    }
+
+    @Override
+    public ByteBuffer getAsByteBuffer() {
+        if (size() == 1) {
+            PersistedData data = getArrayItem(0);
+            if (data.isBytes()) {
+                return data.getAsByteBuffer();
+            }
+        }
+        throw new DeserializationException(String.format("Source is not of type bytes array: %s", type.name()));
+    }
+}
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedDataMap.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedDataMap.java
new file mode 100644
index 00000000000..40b63532a4d
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedDataMap.java
@@ -0,0 +1,113 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling.bytebuffer;
+
+import org.terasology.persistence.typeHandling.PersistedData;
+import org.terasology.persistence.typeHandling.PersistedDataArray;
+import org.terasology.persistence.typeHandling.PersistedDataMap;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * ByteBuffer-backed persisted data map representation.
+ * <pre>
+ *     1 byte -  type = 10
+ *     4 byte - size
+ *     8 * size bytes - refmap
+ *     0..n bytes - key and value data.
+ * </pre>
+ * <p>
+ * Use refmap - links to key and value positions. provide almost constant read time.
+ */
+public class ByteBufferPersistedDataMap extends ByteBufferPersistedData implements PersistedDataMap {
+
+    public ByteBufferPersistedDataMap(ByteBuffer byteBuffer) {
+        super(byteBuffer);
+    }
+
+    @Override
+    public boolean has(String name) {
+        return index(name) != -1;
+    }
+
+    @Override
+    public PersistedData get(String name) {
+        return new ByteBufferPersistedData(byteBuffer, index(name));
+    }
+
+    @Override
+    public float getAsFloat(String name) {
+        return get(name).getAsFloat();
+    }
+
+    @Override
+    public int getAsInteger(String name) {
+        return get(name).getAsInteger();
+    }
+
+    @Override
+    public double getAsDouble(String name) {
+        return get(name).getAsDouble();
+    }
+
+    @Override
+    public long getAsLong(String name) {
+        return get(name).getAsLong();
+    }
+
+    @Override
+    public String getAsString(String name) {
+        return get(name).getAsString();
+    }
+
+    @Override
+    public boolean getAsBoolean(String name) {
+        return get(name).getAsBoolean();
+    }
+
+    @Override
+    public PersistedDataMap getAsMap(String name) {
+        return get(name).getAsValueMap();
+    }
+
+    @Override
+    public PersistedDataArray getAsArray(String name) {
+        return get(name).getAsArray();
+    }
+
+    @Override
+    public Set<Map.Entry<String, PersistedData>> entrySet() {
+        int refPositions = position + 1 + 4;
+        int size = size();
+        Map<String, PersistedData> map = new HashMap<>(size);
+        for (int i = 0; i < size; i++) {
+            int keyPos = position + byteBuffer.getInt(refPositions + 8 * i);
+            int valuePos = position + byteBuffer.getInt(refPositions + 8 * i + 4);
+            String key = new ByteBufferPersistedData(byteBuffer, keyPos, BBType.STRING.getCode()).getAsString();
+            PersistedData value = new ByteBufferPersistedData(byteBuffer, valuePos);
+            map.put(key, value);
+        }
+        return map.entrySet();
+    }
+
+    private int size() {
+        return byteBuffer.getInt(position + 1);
+    }
+
+    private int index(String name) {
+        int refPositions = position + 1 + 4;
+        int size = size();
+        for (int i = 0; i < size; i++) {
+            int keyPos = position + byteBuffer.getInt(refPositions + 8 * i);
+            String candidateKey = new ByteBufferPersistedData(byteBuffer, keyPos, BBType.STRING.getCode()).getAsString();
+            if (name.equals(candidateKey)) {
+                return position + byteBuffer.getInt(refPositions + 8 * i + 4);
+            }
+        }
+        return -1;
+    }
+}
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedSerializer.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedSerializer.java
new file mode 100644
index 00000000000..3b0b6f7682d
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/ByteBufferPersistedSerializer.java
@@ -0,0 +1,326 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling.bytebuffer;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import gnu.trove.iterator.TDoubleIterator;
+import gnu.trove.iterator.TFloatIterator;
+import gnu.trove.iterator.TIntIterator;
+import gnu.trove.iterator.TLongIterator;
+import gnu.trove.list.array.TDoubleArrayList;
+import gnu.trove.list.array.TFloatArrayList;
+import gnu.trove.list.array.TIntArrayList;
+import gnu.trove.list.array.TLongArrayList;
+import org.terasology.persistence.typeHandling.PersistedData;
+import org.terasology.persistence.typeHandling.PersistedDataSerializer;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+
+/**
+ * Provide possible serialize data to ByteBuffer.
+ */
+public class ByteBufferPersistedSerializer implements PersistedDataSerializer {
+
+    @Override
+    public PersistedData serialize(String value) {
+        byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + bytes.length);
+        buffer.put(BBType.STRING.getCode());
+        buffer.putInt(bytes.length);
+        buffer.put(bytes);
+        buffer.rewind();
+        return new ByteBufferPersistedData(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(String... values) {
+        return serializeStrings(Arrays.asList(values));
+    }
+
+    @Override
+    public PersistedData serializeStrings(Iterable<String> value) {
+        List<ByteBufferPersistedData> buffers = Lists.newArrayList();
+        int byteSize = 0;
+        int size = 0;
+        for (String str : value) {
+            ByteBufferPersistedData serialize = (ByteBufferPersistedData) serialize(str);
+            byteSize += serialize.getData().array().length;
+            size++;
+            buffers.add(serialize);
+        }
+        if (size == 1) {
+            return buffers.get(0);
+        }
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.SIZE_HEADER + byteSize);
+        buffer.put(BBType.STRING_ARRAY.getCode());
+        buffer.putInt(size);
+        for (ByteBufferPersistedData data : buffers) {
+            data.getData().position(1); // remove header of string.
+            buffer.put(data.getData());
+        }
+        buffer.rewind();
+        return new ByteBufferPersistedDataArray(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(float value) {
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.FLOAT_SIZE);
+        buffer.put(BBType.FLOAT.getCode());
+        buffer.putFloat(value);
+        buffer.rewind();
+        return new ByteBufferPersistedData(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(float... values) {
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + values.length * BBConsts.FLOAT_SIZE);
+        buffer.put(BBType.FLOAT_ARRAY.getCode());
+        buffer.putInt(values.length);
+        for (float value : values) {
+            buffer.putFloat(value);
+        }
+        buffer.rewind();
+        return new ByteBufferPersistedDataArray(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(TFloatIterator value) {
+        TFloatArrayList data = new TFloatArrayList();
+        while (value.hasNext()) {
+            data.add(value.next());
+        }
+        return serialize(data.toArray());
+    }
+
+    @Override
+    public PersistedData serialize(int value) {
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.INT_SIZE);
+        buffer.put(BBType.INTEGER.getCode());
+        buffer.putInt(value);
+        buffer.rewind();
+        return new ByteBufferPersistedData(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(int... values) {
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + values.length * BBConsts.INT_SIZE);
+        buffer.put(BBType.INTEGER_ARRAY.getCode());
+        buffer.putInt(values.length);
+        for (int value : values) {
+            buffer.putInt(value);
+        }
+        buffer.rewind();
+        return new ByteBufferPersistedDataArray(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(TIntIterator value) {
+        TIntArrayList data = new TIntArrayList();
+        while (value.hasNext()) {
+            data.add(value.next());
+        }
+        return serialize(data.toArray());
+    }
+
+    @Override
+    public PersistedData serialize(long value) {
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.LONG_SIZE);
+        buffer.put(BBType.LONG.getCode());
+        buffer.putLong(value);
+        buffer.rewind();
+        return new ByteBufferPersistedData(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(long... values) {
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + values.length * BBConsts.LONG_SIZE);
+        buffer.put(BBType.LONG_ARRAY.getCode());
+        buffer.putInt(values.length);
+        for (long value : values) {
+            buffer.putLong(value);
+        }
+        buffer.rewind();
+        return new ByteBufferPersistedDataArray(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(TLongIterator value) {
+        TLongArrayList data = new TLongArrayList();
+        while (value.hasNext()) {
+            data.add(value.next());
+        }
+        return serialize(data.toArray());
+    }
+
+    @Override
+    public PersistedData serialize(boolean value) {
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.BYTE_SIZE);
+        buffer.put(BBType.BOOLEAN.getCode());
+        buffer.put(value ? (byte) 1 : (byte) 0);
+        buffer.rewind();
+        return new ByteBufferPersistedData(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(boolean... values) {
+        int size = values.length;
+        int sizeInBytes = size % 8 + 1;
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + sizeInBytes);
+        buffer.put(BBType.BOOLEAN_ARRAY.getCode());
+        buffer.putInt(size);
+        for (int i = 0; i < sizeInBytes; i++) {
+            byte valueByte = 0;
+            for (int bi = 0; bi < 8; bi++) {
+                if (i * 8 + bi >= size) {
+                    break;
+                }
+                boolean b = values[i * 8 + bi];
+                valueByte |= (b ? 1 : 0) << bi;
+            }
+            buffer.put(valueByte);
+        }
+        buffer.rewind();
+        return new ByteBufferPersistedDataArray(buffer);
+
+    }
+
+    @Override
+    public PersistedData serialize(double value) {
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.DOUBLE_SIZE);
+        buffer.put(BBType.DOUBLE.getCode());
+        buffer.putDouble(value);
+        buffer.rewind();
+        return new ByteBufferPersistedData(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(double... values) {
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + values.length * BBConsts.DOUBLE_SIZE);
+        buffer.put(BBType.DOUBLE_ARRAY.getCode());
+        buffer.putInt(values.length);
+        for (double value : values) {
+            buffer.putDouble(value);
+        }
+        buffer.rewind();
+        return new ByteBufferPersistedDataArray(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(TDoubleIterator value) {
+        TDoubleArrayList data = new TDoubleArrayList();
+        while (value.hasNext()) {
+            data.add(value.next());
+        }
+        return serialize(data.toArray());
+    }
+
+    @Override
+    public PersistedData serialize(byte[] value) {
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + value.length);
+        buffer.put(BBType.BYTES.getCode());
+        buffer.putInt(value.length);
+        buffer.put(value);
+        buffer.rewind();
+        return new ByteBufferPersistedData(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(ByteBuffer value) {
+        int size = value.array().length;
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + size);
+        buffer.put(BBType.BYTEBUFFER.getCode());
+        buffer.putInt(size);
+        buffer.put(value);
+        buffer.rewind();
+        return new ByteBufferPersistedData(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(PersistedData... values) {
+        int bytes = 0;
+        int[] sizes = new int[values.length];
+        for (int i = 0; i < values.length; i++) {
+            PersistedData data = values[i];
+            ByteBufferPersistedData dataBytes = (ByteBufferPersistedData) data;
+            int length = dataBytes.byteBuffer.array().length;
+            bytes += length;
+            sizes[i] = length;
+        }
+        int refSize = BBConsts.INT_SIZE;
+        ByteBuffer buffer = ByteBuffer.allocate(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + values.length * refSize + bytes);
+        buffer.put(BBType.VALUE_ARRAY.getCode());
+        buffer.putInt(values.length);
+        // Write data sizes
+        for (int size : sizes) {
+            buffer.putInt(size);
+        }
+        for (PersistedData data : values) {
+            ByteBufferPersistedData dataBytes = (ByteBufferPersistedData) data;
+            dataBytes.byteBuffer.rewind();
+            buffer.put(dataBytes.byteBuffer);
+        }
+        buffer.rewind();
+        return new ByteBufferPersistedDataArray(buffer);
+    }
+
+    @Override
+    public PersistedData serialize(Iterable<PersistedData> data) {
+        return serialize(Iterables.toArray(data, PersistedData.class));
+    }
+
+    @Override
+    public PersistedData serialize(Map<String, PersistedData> data) {
+        int entryCount = data.size();
+        int size = 0;
+        List<PersistedData> keys = new ArrayList<>(entryCount);
+        List<PersistedData> values = new ArrayList<>(entryCount);
+        for (Map.Entry<String, PersistedData> entry : data.entrySet()) {
+            PersistedData serialize = serialize(entry.getKey());
+            ByteBuffer keyBuffer = ((ByteBufferPersistedData) serialize).byteBuffer;
+            keyBuffer.position(1); // skip header
+            size += keyBuffer.remaining();
+            keys.add(serialize);
+            size += ((ByteBufferPersistedData) entry.getValue()).byteBuffer.array().length;
+            values.add(entry.getValue());
+        }
+        int oneRefSize = BBConsts.INT_SIZE;
+        int refsSize = oneRefSize * 2 * entryCount;
+        ByteBuffer buffer = ByteBuffer.allocate(1 + 4 + refsSize + size);
+        buffer.put(BBType.VALUEMAP.getCode());
+        buffer.putInt(entryCount);
+        int dataPosition = 1 + 4 + refsSize;
+        buffer.position(dataPosition);
+        for (int i = 0; i < keys.size(); i++) {
+            ByteBufferPersistedData key = (ByteBufferPersistedData) keys.get(i);
+            int keySize = key.byteBuffer.remaining();
+            buffer.put(key.byteBuffer);
+            buffer.putInt(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + i * oneRefSize * 2, dataPosition);
+            dataPosition += keySize;
+        }
+        for (int i = 0; i < values.size(); i++) {
+            ByteBufferPersistedData value = (ByteBufferPersistedData) values.get(i);
+            int keySize = value.byteBuffer.array().length;
+            buffer.put(value.byteBuffer.array());
+            buffer.putInt(BBConsts.TYPE_HEADER + BBConsts.SIZE_HEADER + i * oneRefSize * 2 + oneRefSize, dataPosition);
+            dataPosition += keySize;
+        }
+        buffer.rewind();
+        return new ByteBufferPersistedDataMap(buffer);
+    }
+
+    @Override
+    public PersistedData serializeNull() {
+        ByteBuffer buffer = ByteBuffer.allocate(1)
+                .put(BBType.NULL.getCode());
+        buffer.rewind();
+        return new ByteBufferPersistedData(buffer);
+    }
+}
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/package-info.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/package-info.java
new file mode 100644
index 00000000000..9b0a0750405
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/bytebuffer/package-info.java
@@ -0,0 +1,49 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+/**
+ * ByteBuffer serializer provide possible to read and write bytebuffer directly.
+ * <p>
+ * <pre>
+ *         Format:
+ *         Types(BBType):
+ *            NULL(0) - size = 1 byte, format = type(byte value = 0)
+ *            BOOLEAN(1) - size = 2 bytes, format = type(byte value = 1) + value(1 byte). ( yeah a bit not optimal)
+ *            FLOAT(2) - size = 5 bytes, format = type(byte value = 2) + value(1 float = 4 bytes)
+ *            DOUBLE(3) - size = 9 bytes, format = type(byte value = 3) + value(1 double = 8 bytes)
+ *            LONG(4) -  size = 9 bytes, format = type(byte value = 4) + value(1 long = 8 bytes)
+ *            INTEGER(5) - size = 5 bytes, format = type(byte value = 5) + value(1 int = 4 bytes)
+ *            STRING(6) - size = 5..n bytes, format =  type(byte value = 6) + size(1 int = 4 bytes) + value($size bytes)
+ *            BYTES(7) - size = 5..n bytes, format =  type(byte value = 7) + size(1 int = 4 bytes) + value($size bytes)
+ *            BYTEBUFFER(8) - size = 5..n bytes, format =  type(byte value = 8) + size(1 int = 4 bytes) + value($size bytes)
+ *            BOOLEAN(9) - size = 5..n bytes.
+ *                       format = type(byte value = 9) + size(1 int = 4 bytes) + data(($size % 8 + 1) bytes)
+ *            FLOAT(10) - size = 5 .. n bytes.
+ *                       format = type(byte value = 10) + size(1 int = 4 bytes) + data($size floats = $size * 4 bytes)
+ *            DOUBLE(11) - size = 5 .. n bytes.
+ *                       format = type(byte value = 11) + size(1 int = 4 bytes) + data($size double = $size * 8 bytes)
+ *            LONG(12) - size = 5 .. n bytes.
+ *                       format = type(byte value = 12) + size(1 int = 4 bytes) + data($size longs = $size * 8 bytes)
+ *            INTEGER(13) - size = 5 .. n bytes.
+ *                       format = type(byte value = 13) + size(1 int = 4 bytes) + data($size ints = $size * 4 bytes)
+ *            STRING(14) - size = 5 .. n bytes.
+ *                       format = type(byte value = 14) +
+ *                                size(1 int = 4 bytes) +
+ *                                sizeArray(format = STRING(6).size = 1 int * $size = 4 byte * $size) +
+ *                                stringdata(format = (STRING(6) - 1 byte($STRING.type)) * $size)
+ *            VALUE(15) - size = 5 .. n bytes.
+ *                       format = type(byte value = 15) +
+ *                                size(1 int = 4 bytes) +
+ *                                sizeArray(format = (any BBType whole size) = 1 int * $size = 4 byte * $size) +
+ *                                stringdata(format = (any BBType) * $size)
+ *            VALUEMAP(16) -
+ *                  size = 5..n bytes.
+ *                  format =  type(byte value = 16) +
+ *                            size(1 int = 4 bytes) +
+ *                            refmap(format = (keyRef(1 int) + valueRef(1 int)) * $size) +
+ *                            keydata(format = (STRING(6) - 1 byte($STRING.type) * $size ) +
+ *                            valuedata(format = (any BBType) * $size)
+ *
+ *     </pre>
+ */
+package org.terasology.persistence.typeHandling.bytebuffer;
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/AbstractPersistedData.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/AbstractPersistedData.java
index e40a230b229..27d08f0e71d 100644
--- a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/AbstractPersistedData.java
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/AbstractPersistedData.java
@@ -7,9 +7,10 @@
 import org.terasology.persistence.typeHandling.PersistedDataArray;
 import org.terasology.persistence.typeHandling.PersistedDataMap;
 
+import java.io.Serializable;
 import java.nio.ByteBuffer;
 
-public abstract class AbstractPersistedData implements PersistedData {
+public abstract class AbstractPersistedData implements PersistedData, Serializable {
 
     @Override
     public String getAsString() {
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/InMemoryPersistedDataSerializer.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/InMemoryPersistedDataSerializer.java
index be640bbb779..8a2e650948c 100644
--- a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/InMemoryPersistedDataSerializer.java
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/InMemoryPersistedDataSerializer.java
@@ -28,7 +28,7 @@
 
 public class InMemoryPersistedDataSerializer implements PersistedDataSerializer {
 
-    public static final PersistedData NULL = new AbstractPersistedData() {
+    public static final AbstractPersistedData NULL = new AbstractPersistedData() {
         @Override
         public boolean isNull() {
             return true;
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/InMemoryReader.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/InMemoryReader.java
new file mode 100644
index 00000000000..4039bbeab97
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/InMemoryReader.java
@@ -0,0 +1,41 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling.inMemory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.terasology.persistence.serializers.PersistedDataReader;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.ObjectInputStream;
+import java.nio.ByteBuffer;
+
+public class InMemoryReader implements PersistedDataReader<AbstractPersistedData> {
+    private static final Logger logger = LoggerFactory.getLogger(InMemoryReader.class);
+
+    @Override
+    public AbstractPersistedData read(InputStream inputStream) throws IOException {
+        ObjectInputStream ois = new ObjectInputStream(inputStream);
+        try {
+            return (AbstractPersistedData) ois.readObject();
+        } catch (ClassNotFoundException e) {
+            logger.error("Cannot read to inputStream");
+            return InMemoryPersistedDataSerializer.NULL;
+        }
+    }
+
+    @Override
+    public AbstractPersistedData read(byte[] byteBuffer) throws IOException {
+        return read(new ByteArrayInputStream(byteBuffer));
+    }
+
+    @Override
+    public AbstractPersistedData read(ByteBuffer byteBuffer) throws IOException {
+        byte[] bytes = new byte[byteBuffer.remaining()];
+        byteBuffer.get(bytes);
+        return read(bytes);
+    }
+}
diff --git a/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/InMemoryWriter.java b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/InMemoryWriter.java
new file mode 100644
index 00000000000..2800be2f757
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/main/java/org/terasology/persistence/typeHandling/inMemory/InMemoryWriter.java
@@ -0,0 +1,40 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling.inMemory;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.terasology.persistence.serializers.PersistedDataWriter;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+public class InMemoryWriter implements PersistedDataWriter<AbstractPersistedData> {
+    private static final  Logger logger = LoggerFactory.getLogger(InMemoryWriter.class);
+
+    @Override
+    public byte[] writeBytes(AbstractPersistedData data) {
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        try {
+            writeTo(data, baos);
+        } catch (IOException e) {
+            logger.error("Cannot writeBytes", e);
+        }
+        return baos.toByteArray();
+    }
+
+    @Override
+    public void writeTo(AbstractPersistedData data, OutputStream outputStream) throws IOException {
+        ObjectOutputStream oos = new ObjectOutputStream(outputStream);
+        oos.writeObject(data);
+    }
+
+    @Override
+    public void writeTo(AbstractPersistedData data, ByteBuffer byteBuffer) throws IOException {
+        byteBuffer.put(writeBytes(data));
+    }
+}
diff --git a/subsystems/TypeHandlerLibrary/src/test/java/org/terasology/persistence/typeHandling/ByteBufferSerializerTest.java b/subsystems/TypeHandlerLibrary/src/test/java/org/terasology/persistence/typeHandling/ByteBufferSerializerTest.java
new file mode 100644
index 00000000000..d3c53f63985
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/test/java/org/terasology/persistence/typeHandling/ByteBufferSerializerTest.java
@@ -0,0 +1,571 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling;
+
+import gnu.trove.list.array.TDoubleArrayList;
+import gnu.trove.list.array.TFloatArrayList;
+import gnu.trove.list.array.TIntArrayList;
+import gnu.trove.list.array.TLongArrayList;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.internal.util.collections.Sets;
+import org.terasology.persistence.typeHandling.bytebuffer.ByteBufferPersistedData;
+import org.terasology.persistence.typeHandling.bytebuffer.ByteBufferPersistedDataArray;
+import org.terasology.persistence.typeHandling.bytebuffer.ByteBufferPersistedSerializer;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+
+class ByteBufferSerializerTest {
+    private static PersistedDataSerializer serializer = new ByteBufferPersistedSerializer();
+
+    public static Stream<Arguments> types() {
+        return Stream.of(
+                Arguments.of(serializer.serialize(1),
+                        Sets.newSet(TypeGetter.INTEGER, TypeGetter.LONG, TypeGetter.FLOAT, TypeGetter.DOUBLE)),
+                Arguments.of(serializer.serialize(1L),
+                        Sets.newSet(TypeGetter.INTEGER, TypeGetter.LONG, TypeGetter.FLOAT, TypeGetter.DOUBLE)),
+                Arguments.of(serializer.serialize(1F),
+                        Sets.newSet(TypeGetter.INTEGER, TypeGetter.LONG, TypeGetter.FLOAT, TypeGetter.DOUBLE)),
+                Arguments.of(serializer.serialize(1D),
+                        Sets.newSet(TypeGetter.INTEGER, TypeGetter.LONG, TypeGetter.FLOAT, TypeGetter.DOUBLE)),
+
+                Arguments.of(serializer.serialize("foo"), // TODO
+                        Sets.newSet(TypeGetter.STRING)),
+
+                Arguments.of(serializer.serialize(new byte[]{(byte) 0xFF}),
+                        Sets.newSet(TypeGetter.BYTE_BUFFER, TypeGetter.BYTES)),
+                Arguments.of(serializer.serialize(ByteBuffer.wrap(new byte[]{(byte) 0xFF})),
+                        Sets.newSet(TypeGetter.BYTES, TypeGetter.BYTE_BUFFER)),
+                Arguments.of(serializer.serialize(true),
+                        Sets.newSet(TypeGetter.BOOLEAN))
+        );
+    }
+
+    @Test
+    void serializeString() {
+        PersistedData data = serializer.serialize("foo");
+
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getClass());
+        Assertions.assertTrue(data.isString());
+        Assertions.assertEquals("foo", data.getAsString());
+
+        Assertions.assertFalse(data.isArray());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isNumber());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isBytes());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsArray);
+
+        Assertions.assertThrows(DeserializationException.class, data::getAsByteBuffer);
+        Assertions.assertThrows(DeserializationException.class, data::getAsBytes);
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsBoolean);
+        Assertions.assertThrows(ClassCastException.class, data::getAsInteger);
+        Assertions.assertThrows(ClassCastException.class, data::getAsLong);
+        Assertions.assertThrows(ClassCastException.class, data::getAsFloat);
+        Assertions.assertThrows(ClassCastException.class, data::getAsDouble);
+    }
+
+    @Test
+    void serializeStrings() {
+        PersistedData data = serializer.serialize("foo", "bar");
+        Assertions.assertEquals(ByteBufferPersistedDataArray.class, data.getClass());
+
+        Assertions.assertTrue(data.isArray());
+        Assertions.assertEquals("foo", data.getAsArray().getArrayItem(0).getAsString());
+
+        Assertions.assertTrue(data.getAsArray().isStringArray());
+        Assertions.assertEquals("foo", data.getAsArray().getAsStringArray().get(0));
+
+        Assertions.assertTrue(data.isArray());
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isBytes());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(IllegalStateException.class, data::getAsString);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+
+        Assertions.assertThrows(DeserializationException.class, data::getAsByteBuffer);
+        Assertions.assertThrows(DeserializationException.class, data::getAsBytes);
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsBoolean);
+        Assertions.assertThrows(ClassCastException.class, data::getAsInteger);
+        Assertions.assertThrows(ClassCastException.class, data::getAsLong);
+        Assertions.assertThrows(ClassCastException.class, data::getAsFloat);
+        Assertions.assertThrows(ClassCastException.class, data::getAsDouble);
+    }
+
+    //TODO remove it
+    public void template(PersistedData data) {
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isArray());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isNumber());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isBytes());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsArray);
+
+        Assertions.assertThrows(DeserializationException.class, data::getAsByteBuffer);
+        Assertions.assertThrows(DeserializationException.class, data::getAsBytes);
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsString);
+        Assertions.assertThrows(ClassCastException.class, data::getAsBoolean);
+        Assertions.assertThrows(ClassCastException.class, data::getAsInteger);
+        Assertions.assertThrows(ClassCastException.class, data::getAsLong);
+        Assertions.assertThrows(ClassCastException.class, data::getAsFloat);
+        Assertions.assertThrows(ClassCastException.class, data::getAsDouble);
+    }
+
+    @Test
+    void serializeOneAsStrings() {
+        PersistedData data = serializer.serialize(new String[]{"foo"});
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getClass());
+
+        Assertions.assertEquals("foo", data.getAsString());
+    }
+
+    @Test
+    void serializeStringsIterable() {
+        PersistedData data = serializer.serializeStrings(Arrays.asList("foo", "bar"));
+        Assertions.assertEquals(ByteBufferPersistedDataArray.class, data.getClass());
+
+        Assertions.assertTrue(data.isArray());
+        Assertions.assertEquals("foo", data.getAsArray().getArrayItem(0).getAsString());
+
+        Assertions.assertTrue(data.getAsArray().isStringArray());
+        Assertions.assertEquals("foo", data.getAsArray().getAsStringArray().get(0));
+    }
+
+    @Test
+    void serializeOneAsStringsIterable() {
+        PersistedData data = serializer.serializeStrings(Collections.singleton("foo"));
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getClass());
+
+        Assertions.assertEquals("foo", data.getAsString());
+    }
+
+    @Test
+    void serializeFloat() {
+        PersistedData data = serializer.serialize(1f);
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getClass());
+        checkIsNumber(data);
+    }
+
+    @Test
+    void serializeFloats() {
+        PersistedData data = serializer.serialize(new float[]{1F});
+        checkNumberArray(data, ByteBufferPersistedDataArray.class, ByteBufferPersistedData.class);
+    }
+
+    @Test
+    void serializeTFloatIterator() {
+        PersistedData data = serializer.serialize(TFloatArrayList.wrap(new float[]{1F}).iterator());
+        checkNumberArray(data, ByteBufferPersistedDataArray.class, ByteBufferPersistedData.class);
+    }
+
+    @Test
+    void serializeInt() {
+        PersistedData data = serializer.serialize(1);
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getClass());
+
+        checkIsNumber(data);
+    }
+
+    @Test
+    void serializeInts() {
+        PersistedData data = serializer.serialize(new int[]{1});
+        checkNumberArray(data, ByteBufferPersistedDataArray.class, ByteBufferPersistedData.class);
+    }
+
+    @Test
+    void serializeTIntIterator() {
+        PersistedData data = serializer.serialize(TIntArrayList.wrap(new int[]{1}).iterator());
+        checkNumberArray(data, ByteBufferPersistedDataArray.class, ByteBufferPersistedData.class);
+    }
+
+    @Test
+    void serializeLong() {
+        PersistedData data = serializer.serialize(1L);
+
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getClass());
+
+        checkIsNumber(data);
+    }
+
+    @Test
+    void serializeLongs() {
+        PersistedData data = serializer.serialize(new long[]{1L});
+        checkNumberArray(data, ByteBufferPersistedDataArray.class, ByteBufferPersistedData.class);
+    }
+
+    @Test
+    void serializeTLongIterator() {
+        PersistedData data = serializer.serialize(TLongArrayList.wrap(new long[]{1L}).iterator());
+        checkNumberArray(data, ByteBufferPersistedDataArray.class, ByteBufferPersistedData.class);
+    }
+
+    @Test
+    void serializeBoolean() {
+        boolean value = true;
+        PersistedData data = serializer.serialize(value);
+
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getClass());
+
+        Assertions.assertTrue(data.isBoolean());
+        Assertions.assertEquals(value, data.getAsBoolean());
+
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isArray());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isNumber());
+        Assertions.assertFalse(data.isBytes());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsArray);
+
+        Assertions.assertThrows(DeserializationException.class, data::getAsByteBuffer);
+        Assertions.assertThrows(DeserializationException.class, data::getAsBytes);
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsString);
+        Assertions.assertThrows(ClassCastException.class, data::getAsInteger);
+        Assertions.assertThrows(ClassCastException.class, data::getAsLong);
+        Assertions.assertThrows(ClassCastException.class, data::getAsFloat);
+        Assertions.assertThrows(ClassCastException.class, data::getAsDouble);
+    }
+
+    @Test
+    void serializeBooleans() {
+        PersistedData data = serializer.serialize(new boolean[]{true});
+
+        Assertions.assertEquals(ByteBufferPersistedDataArray.class, data.getClass());
+
+        Assertions.assertTrue(data.isArray());
+
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isBytes());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsString);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+
+        Assertions.assertThrows(DeserializationException.class, data::getAsByteBuffer);
+        Assertions.assertThrows(DeserializationException.class, data::getAsBytes);
+
+        Assertions.assertTrue(data.getAsArray().isBooleanArray());
+
+        Assertions.assertTrue(data.getAsBoolean());
+        Assertions.assertArrayEquals(new boolean[]{true}, data.getAsArray().getAsBooleanArray());
+
+
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getAsArray().getArrayItem(0).getClass());
+    }
+
+
+    @Test
+    void serializeBooleansComplicated() {
+        boolean[] value = {true, true, true, false, true, false, true, false, true, true, true, false, true};
+        PersistedData data = serializer.serialize(value);
+
+        Assertions.assertEquals(ByteBufferPersistedDataArray.class, data.getClass());
+
+        Assertions.assertTrue(data.isArray());
+
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isBytes());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsString);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+
+        Assertions.assertThrows(DeserializationException.class, data::getAsByteBuffer);
+        Assertions.assertThrows(DeserializationException.class, data::getAsBytes);
+
+        Assertions.assertThrows(IllegalStateException.class, data::getAsBoolean);
+
+        Assertions.assertArrayEquals(value, data.getAsArray().getAsBooleanArray());
+
+        Assertions.assertTrue(data.getAsArray().isBooleanArray());
+
+
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getAsArray().getArrayItem(0).getClass());
+    }
+
+    @Test
+    void serializeDouble() {
+        PersistedData data = serializer.serialize(1D);
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getClass());
+
+        checkIsNumber(data);
+    }
+
+    @Test
+    void serializeDoubles() {
+        PersistedData data = serializer.serialize(new double[]{1D});
+        checkNumberArray(data, ByteBufferPersistedDataArray.class, ByteBufferPersistedData.class);
+    }
+
+    @Test
+    void serializeTDoubleIterator() {
+        PersistedData data = serializer.serialize(TDoubleArrayList.wrap(new double[]{1D}).iterator());
+        checkNumberArray(data, ByteBufferPersistedDataArray.class, ByteBufferPersistedData.class);
+    }
+
+    @Test
+    void serializeBytes() {
+        byte[] value = {(byte) 0xFF};
+        PersistedData data = serializer.serialize(value);
+
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getClass());
+        Assertions.assertTrue(data.isBytes());
+        Assertions.assertArrayEquals(value, data.getAsBytes());
+        Assertions.assertEquals(ByteBuffer.wrap(value), data.getAsByteBuffer());
+
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isArray());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isNumber());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsArray);
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsString);
+        Assertions.assertThrows(ClassCastException.class, data::getAsBoolean);
+        Assertions.assertThrows(ClassCastException.class, data::getAsInteger);
+        Assertions.assertThrows(ClassCastException.class, data::getAsLong);
+        Assertions.assertThrows(ClassCastException.class, data::getAsFloat);
+        Assertions.assertThrows(ClassCastException.class, data::getAsDouble);
+    }
+
+    @Test
+    void serializeByteBuffer() {
+        byte[] value = {(byte) 0xFF};
+        PersistedData data = serializer.serialize(ByteBuffer.wrap(value));
+
+        Assertions.assertEquals(ByteBufferPersistedData.class, data.getClass());
+        Assertions.assertTrue(data.isBytes());
+
+        Assertions.assertArrayEquals(value, data.getAsBytes());
+        Assertions.assertEquals(ByteBuffer.wrap(value), data.getAsByteBuffer());
+
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isArray());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isNumber());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsArray);
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsString);
+        Assertions.assertThrows(ClassCastException.class, data::getAsBoolean);
+        Assertions.assertThrows(ClassCastException.class, data::getAsInteger);
+        Assertions.assertThrows(ClassCastException.class, data::getAsLong);
+        Assertions.assertThrows(ClassCastException.class, data::getAsFloat);
+        Assertions.assertThrows(ClassCastException.class, data::getAsDouble);
+    }
+
+    @ParameterizedTest(name = "{1}")
+    @MethodSource("types")
+    void serializePersistedDatas(PersistedData entry, Set<TypeGetter> typeGetters) {
+        PersistedData data = serializer.serialize(entry);
+        checkValueArray(data, entry, typeGetters);
+    }
+
+    @ParameterizedTest(name = "{1}")
+    @MethodSource("types")
+    void serializeIterablePersistedData(PersistedData entry, Set<TypeGetter> typeGetters) {
+        PersistedData data = serializer.serialize(Collections.singletonList(entry));
+        checkValueArray(data, entry, typeGetters);
+    }
+
+    @Test
+    void serializeMapStringPersistedData() {
+
+    }
+
+    @Test
+    void serializeNull() {
+        PersistedData data = serializer.serializeNull();
+
+        Assertions.assertTrue(data.isNull());
+
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isArray());
+        Assertions.assertFalse(data.isNumber());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isBytes());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsArray);
+
+        Assertions.assertThrows(DeserializationException.class, data::getAsByteBuffer);
+        Assertions.assertThrows(DeserializationException.class, data::getAsBytes);
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsString);
+        Assertions.assertThrows(ClassCastException.class, data::getAsBoolean);
+        Assertions.assertThrows(ClassCastException.class, data::getAsInteger);
+        Assertions.assertThrows(ClassCastException.class, data::getAsLong);
+        Assertions.assertThrows(ClassCastException.class, data::getAsFloat);
+        Assertions.assertThrows(ClassCastException.class, data::getAsDouble);
+    }
+
+    private void checkNumberArray(PersistedData data, Class<?> arrayType,
+                                  Class<?> entryType) {
+        Assertions.assertEquals(arrayType, data.getClass());
+
+        Assertions.assertTrue(data.isArray());
+
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isBytes());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsString);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+
+        Assertions.assertThrows(DeserializationException.class, data::getAsByteBuffer);
+        Assertions.assertThrows(DeserializationException.class, data::getAsBytes);
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsBoolean);
+
+        Assertions.assertEquals(1, data.getAsInteger());
+        Assertions.assertEquals(1L, data.getAsLong());
+        Assertions.assertEquals(1F, data.getAsFloat());
+        Assertions.assertEquals(1D, data.getAsDouble());
+
+        Assertions.assertEquals(entryType, data.getAsArray().getArrayItem(0).getClass());
+
+        Assertions.assertEquals(1, data.getAsArray().getAsIntegerArray().get(0));
+        Assertions.assertEquals(1L, data.getAsArray().getAsLongArray().get(0));
+        Assertions.assertEquals(1F, data.getAsArray().getAsFloatArray().get(0));
+        Assertions.assertEquals(1D, data.getAsArray().getAsDoubleArray().get(0));
+    }
+
+    private void checkIsNumber(PersistedData data) {
+        Assertions.assertTrue(data.isNumber());
+
+        Assertions.assertEquals(1, data.getAsInteger());
+        Assertions.assertEquals(1L, data.getAsLong());
+        Assertions.assertEquals(1F, data.getAsFloat());
+        Assertions.assertEquals(1D, data.getAsDouble());
+
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isArray());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isBytes());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+        Assertions.assertThrows(IllegalStateException.class, data::getAsArray);
+
+        Assertions.assertThrows(DeserializationException.class, data::getAsByteBuffer);
+        Assertions.assertThrows(DeserializationException.class, data::getAsBytes);
+
+        Assertions.assertThrows(ClassCastException.class, data::getAsString);
+        Assertions.assertThrows(ClassCastException.class, data::getAsBoolean);
+    }
+
+    private void checkValueArray(PersistedData data, PersistedData entry, Set<TypeGetter> typeGetters) {
+        Assertions.assertEquals(ByteBufferPersistedDataArray.class, data.getClass());
+
+//        Assertions.assertEquals(entry, data.getAsArray().getArrayItem(0));
+        typeGetters.forEach(typeGetter -> {
+                    Object expected = typeGetter.getGetter().apply(entry);
+                    Object actual = typeGetter.getGetter().apply(data);
+                    if (typeGetter == TypeGetter.BYTES) {
+                        Assertions.assertArrayEquals((byte[]) expected, (byte[]) actual);
+                    } else {
+                        Assertions.assertEquals(expected, actual);
+                    }
+                }
+        );
+
+
+        Assertions.assertFalse(data.isString());
+        Assertions.assertFalse(data.isNull());
+        Assertions.assertFalse(data.isNumber());
+        Assertions.assertFalse(data.isBoolean());
+        Assertions.assertFalse(data.isBytes());
+        Assertions.assertFalse(data.isValueMap());
+
+        Assertions.assertThrows(IllegalStateException.class, data::getAsValueMap);
+
+        Set<TypeGetter> deserializationExceptionGetters = Sets.newSet(
+                TypeGetter.BYTE_BUFFER,
+                TypeGetter.BYTES
+        );
+        deserializationExceptionGetters.stream()
+                .filter(f -> !typeGetters.contains(f))
+                .map(TypeGetter::getGetter)
+                .map(f -> (Executable) () -> f.apply(data))
+                .forEach(e ->
+                        Assertions.assertThrows(DeserializationException.class, e)
+                );
+
+        Set<TypeGetter> classCastExceptionGetters = Sets.newSet(
+                TypeGetter.BOOLEAN,
+                TypeGetter.STRING,
+                TypeGetter.INTEGER,
+                TypeGetter.LONG,
+                TypeGetter.FLOAT,
+                TypeGetter.DOUBLE
+        );
+        classCastExceptionGetters.stream().filter(f -> !typeGetters.contains(f))
+                .map(TypeGetter::getGetter)
+                .map(f -> (Executable) () -> f.apply(data))
+                .forEach(e ->
+                        Assertions.assertThrows(ClassCastException.class, e)
+                );
+    }
+
+    private enum TypeGetter {
+        STRING(PersistedData::getAsString),
+        BOOLEAN(PersistedData::getAsBoolean),
+        INTEGER(PersistedData::getAsInteger),
+        LONG(PersistedData::getAsLong),
+        FLOAT(PersistedData::getAsFloat),
+        DOUBLE(PersistedData::getAsDouble),
+        BYTE_BUFFER(PersistedData::getAsByteBuffer),
+        BYTES(PersistedData::getAsBytes);
+
+        private final Function<PersistedData, Object> getter;
+
+        TypeGetter(Function<PersistedData, Object> getter) {
+            this.getter = getter;
+        }
+
+        public Function<PersistedData, Object> getGetter() {
+            return getter;
+        }
+    }
+}
diff --git a/subsystems/TypeHandlerLibrary/src/test/java/org/terasology/persistence/typeHandling/FullSerializationDeserializationTest.java b/subsystems/TypeHandlerLibrary/src/test/java/org/terasology/persistence/typeHandling/FullSerializationDeserializationTest.java
new file mode 100644
index 00000000000..4a0181646a3
--- /dev/null
+++ b/subsystems/TypeHandlerLibrary/src/test/java/org/terasology/persistence/typeHandling/FullSerializationDeserializationTest.java
@@ -0,0 +1,319 @@
+// Copyright 2021 The Terasology Foundation
+// SPDX-License-Identifier: Apache-2.0
+
+package org.terasology.persistence.typeHandling;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import com.google.common.collect.Streams;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.reflections.Reflections;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.terasology.persistence.serializers.Serializer;
+import org.terasology.persistence.typeHandling.bytebuffer.ByteBufferDataReader;
+import org.terasology.persistence.typeHandling.bytebuffer.ByteBufferDataWriter;
+import org.terasology.persistence.typeHandling.bytebuffer.ByteBufferPersistedSerializer;
+import org.terasology.persistence.typeHandling.inMemory.InMemoryPersistedDataSerializer;
+import org.terasology.persistence.typeHandling.inMemory.InMemoryReader;
+import org.terasology.persistence.typeHandling.inMemory.InMemoryWriter;
+import org.terasology.reflection.TypeInfo;
+
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Stream;
+
+public class FullSerializationDeserializationTest {
+
+    private static final Logger logger = LoggerFactory.getLogger(FullSerializationDeserializationTest.class);
+
+    private static Reflections reflections;
+    private static TypeHandlerLibrary typeHandlerLibrary;
+
+    @BeforeAll
+    public static void setup() {
+        reflections = new Reflections(TypeHandlerLibraryTest.class.getClassLoader());
+        typeHandlerLibrary = new TypeHandlerLibrary(reflections);
+        TypeHandlerLibrary.populateBuiltInHandlers(typeHandlerLibrary);
+
+    }
+
+    private static Stream<Arguments> values() {
+        return Stream.of(
+                value(true),
+                value((byte) 1),
+//                value((short) 1),
+                value(1),
+                value(1L),
+                value(1F),
+                value(1D),
+                value('c'),
+                value("string"),
+                value(Locale.ENGLISH),
+                value(new SampleClass(
+                        "className",
+                        -1,
+                        new SampleClass2("child"),
+                        Lists.newArrayList(
+                                new SampleClass2("child1"),
+                                new SampleClass2("child2")
+                        )
+                ))
+        );
+    }
+
+    private static Stream<Arguments> collections() {
+        return Stream.of(
+                Arguments.of(new TypeInfo<List<String>>() {
+                }, Lists.newArrayList("String1", "String2")),
+                Arguments.of(new TypeInfo<Set<String>>() {
+                }, Sets.newHashSet("String1", "String2")),
+                Arguments.of(new TypeInfo<EnumSet<SampleEnum>>() {
+                }, EnumSet.of(SampleEnum.ONE)),
+                Arguments.of(new TypeInfo<EnumSet<SampleEnum>>() {
+                }, EnumSet.of(SampleEnum.ONE, SampleEnum.THREE)),
+                Arguments.of(new TypeInfo<Map<String, String>>() {
+                }, new HashMap<String, String>() {{
+                    put("key1", "value1");
+                }})
+//                Arguments.arguments(new TypeInfo<Map<String, List<Map<String, SampleEnum>>>>() {
+//                }, new HashMap<String, List<Map<String, SampleEnum>>>() {{
+//                    put("key1", Lists.newArrayList(new HashMap() {{
+//                        put("innerKey", SampleEnum.TWO);
+//                    }}));
+//                }}) Runtime delegate type handler fail it
+        );
+    }
+
+    private static Stream<Arguments> serializers() {
+        return Stream.of(
+                Arguments.of(
+                        new Serializer(typeHandlerLibrary,
+                                new ByteBufferPersistedSerializer(),
+                                new ByteBufferDataWriter(),
+                                new ByteBufferDataReader())),
+                Arguments.of(
+                        new Serializer(typeHandlerLibrary,
+                                new InMemoryPersistedDataSerializer(),
+                                new InMemoryWriter(),
+                                new InMemoryReader()))
+        );
+    }
+
+    private static Stream<Arguments> product() {
+        return serializers()
+                .flatMap(s -> Streams.concat(values(), collections())
+                        .map(v -> Arguments.of(s.get()[0], v.get()[0], v.get()[1]))
+                );
+    }
+
+    private static Arguments value(Object value) {
+        return value(value.getClass(), value);
+    }
+
+    private static Arguments value(Class clazz, Object value) {
+        return Arguments.of(TypeInfo.of(clazz), value);
+    }
+
+
+    @MethodSource("product")
+    @ParameterizedTest(name = "{0} : {1} : {2}")
+    <T> void test(Serializer<PersistedData> serializer, TypeInfo<T> type, T value) {
+        Optional<byte[]> serialized = serializer.serialize(value, type);
+        Assertions.assertTrue(serialized.isPresent(), String.format("Serializer didn't serialize type %s", type));
+        byte[] bytes = serialized.get();
+        logger.info("Size in bytes: {}", bytes.length);
+        Optional<T> deserialized = serializer.deserialize(type, bytes);
+        Assertions.assertTrue(deserialized.isPresent(), String.format("Serializer didn't deserialize type %s", type));
+        Assertions.assertEquals(value, deserialized.get());
+
+    }
+
+    @MethodSource("serializers")
+    @ParameterizedTest(name = "{0} : {displayName}")
+    void testBoolArray(Serializer<PersistedData> serializer) {
+        boolean[] value = new boolean[]{true};
+        TypeInfo<boolean[]> type = TypeInfo.of(boolean[].class);
+        Optional<byte[]> serialized = serializer.serialize(value, type);
+        Assertions.assertTrue(serialized.isPresent(), String.format("Serializer didn't serialize type %s", type));
+        byte[] bytes = serialized.get();
+        logger.info("Size in bytes: {}", bytes.length);
+        Optional<boolean[]> deserialized = serializer.deserialize(type, bytes);
+        Assertions.assertTrue(deserialized.isPresent(), String.format("Serializer didn't deserialize type %s", type));
+        Assertions.assertArrayEquals(value, deserialized.get());
+    }
+
+    @MethodSource("serializers")
+    @ParameterizedTest(name = "{0} : {displayName}")
+    void testByteArray(Serializer<PersistedData> serializer) {
+        byte[] value = new byte[]{(byte) 1};
+        TypeInfo<byte[]> type = TypeInfo.of(byte[].class);
+        Optional<byte[]> serialized = serializer.serialize(value, type);
+        Assertions.assertTrue(serialized.isPresent(), String.format("Serializer didn't serialize type %s", type));
+        byte[] bytes = serialized.get();
+        logger.info("Size in bytes: {}", bytes.length);
+        Optional<byte[]> deserialized = serializer.deserialize(type, bytes);
+        Assertions.assertTrue(deserialized.isPresent(), String.format("Serializer didn't deserialize type %s", type));
+        Assertions.assertArrayEquals(value, deserialized.get());
+    }
+
+    @MethodSource("serializers")
+    @ParameterizedTest(name = "{0} : {displayName}")
+    void testIntArray(Serializer<PersistedData> serializer) {
+        int[] value = new int[]{1};
+        TypeInfo<int[]> type = TypeInfo.of(int[].class);
+        Optional<byte[]> serialized = serializer.serialize(value, type);
+        Assertions.assertTrue(serialized.isPresent(), String.format("Serializer didn't serialize type %s", type));
+        byte[] bytes = serialized.get();
+        logger.info("Size in bytes: {}", bytes.length);
+        Optional<int[]> deserialized = serializer.deserialize(type, bytes);
+        Assertions.assertTrue(deserialized.isPresent(), String.format("Serializer didn't deserialize type %s", type));
+        Assertions.assertArrayEquals(value, deserialized.get());
+    }
+
+    @MethodSource("serializers")
+    @ParameterizedTest(name = "{0} : {displayName}")
+    void testLongArray(Serializer<PersistedData> serializer) {
+        long[] value = new long[]{1L};
+        TypeInfo<long[]> type = TypeInfo.of(long[].class);
+        Optional<byte[]> serialized = serializer.serialize(value, type);
+        Assertions.assertTrue(serialized.isPresent(), String.format("Serializer didn't serialize type %s", type));
+        byte[] bytes = serialized.get();
+        logger.info("Size in bytes: {}", bytes.length);
+        Optional<long[]> deserialized = serializer.deserialize(type, bytes);
+        Assertions.assertTrue(deserialized.isPresent(), String.format("Serializer didn't deserialize type %s", type));
+        Assertions.assertArrayEquals(value, deserialized.get());
+    }
+
+    @MethodSource("serializers")
+    @ParameterizedTest(name = "{0} : {displayName}")
+    void testFloatArray(Serializer<PersistedData> serializer) {
+        float[] value = new float[]{1F};
+        TypeInfo<float[]> type = TypeInfo.of(float[].class);
+        Optional<byte[]> serialized = serializer.serialize(value, type);
+        Assertions.assertTrue(serialized.isPresent(), String.format("Serializer didn't serialize type %s", type));
+        byte[] bytes = serialized.get();
+        logger.info("Size in bytes: {}", bytes.length);
+        Optional<float[]> deserialized = serializer.deserialize(type, bytes);
+        Assertions.assertTrue(deserialized.isPresent(), String.format("Serializer didn't deserialize type %s", type));
+        Assertions.assertArrayEquals(value, deserialized.get());
+    }
+
+    @MethodSource("serializers")
+    @ParameterizedTest(name = "{0} : {displayName}")
+    void testDoubleArray(Serializer<PersistedData> serializer) {
+        double[] value = new double[]{1D};
+        TypeInfo<double[]> type = TypeInfo.of(double[].class);
+        Optional<byte[]> serialized = serializer.serialize(value, type);
+        Assertions.assertTrue(serialized.isPresent(), String.format("Serializer didn't serialize type %s", type));
+        byte[] bytes = serialized.get();
+        logger.info("Size in bytes: {}", bytes.length);
+        Optional<double[]> deserialized = serializer.deserialize(type, bytes);
+        Assertions.assertTrue(deserialized.isPresent(), String.format("Serializer didn't deserialize type %s", type));
+        Assertions.assertArrayEquals(value, deserialized.get());
+    }
+
+    @MethodSource("serializers")
+    @ParameterizedTest(name = "{0} : {displayName}")
+    void testCharArray(Serializer<PersistedData> serializer) {
+        char[] value = new char[]{'c'};
+        TypeInfo<char[]> type = TypeInfo.of(char[].class);
+        Optional<byte[]> serialized = serializer.serialize(value, type);
+        Assertions.assertTrue(serialized.isPresent(), String.format("Serializer didn't serialize type %s", type));
+        byte[] bytes = serialized.get();
+        logger.info("Size in bytes: {}", bytes.length);
+        Optional<char[]> deserialized = serializer.deserialize(type, bytes);
+        Assertions.assertTrue(deserialized.isPresent(), String.format("Serializer didn't deserialize type %s", type));
+        Assertions.assertArrayEquals(value, deserialized.get());
+    }
+
+    @MethodSource("serializers")
+    @ParameterizedTest(name = "{0} : {displayName}")
+    void testObjectArray(Serializer<PersistedData> serializer) {
+        String[] value = new String[]{"String"};
+        TypeInfo<String[]> type = TypeInfo.of(String[].class);
+        Optional<byte[]> serialized = serializer.serialize(value, type);
+        Assertions.assertTrue(serialized.isPresent(), String.format("Serializer didn't serialize type %s", type));
+        byte[] bytes = serialized.get();
+        logger.info("Size in bytes: {}", bytes.length);
+        Optional<String[]> deserialized = serializer.deserialize(type, bytes);
+        Assertions.assertTrue(deserialized.isPresent(), String.format("Serializer didn't deserialize type %s", type));
+        Assertions.assertArrayEquals(value, deserialized.get());
+    }
+
+    public enum SampleEnum {
+        ONE,
+        TWO,
+        THREE
+    }
+
+    public static class SampleClass {
+        String name;
+        int value1;
+        SampleClass2 child;
+        List<SampleClass2> children;
+
+        public SampleClass(String name, int value1, SampleClass2 child, List<SampleClass2> children) {
+            this.name = name;
+            this.value1 = value1;
+            this.child = child;
+            this.children = children;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
+            SampleClass that = (SampleClass) o;
+            return value1 == that.value1
+                    && Objects.equals(name, that.name)
+                    && Objects.equals(child, that.child)
+                    && Objects.equals(children, that.children);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(name, value1, child, children);
+        }
+    }
+
+    public static class SampleClass2 {
+        String name;
+
+        public SampleClass2(String name) {
+            this.name = name;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
+            SampleClass2 that = (SampleClass2) o;
+            return Objects.equals(name, that.name);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(name);
+        }
+    }
+}
+