@Override
public void writeChar(char value) throws KryoException
{
  switch (writeBuffer.remaining()) {
    case 0:
      advanceWriteBuffer();
      writeBuffer.put((byte)(value >>> 8));
      writeBuffer.put((byte)value);
      break;

    case 1:
      writeBuffer.put((byte)(value >>> 8));
      advanceWriteBuffer();
      writeBuffer.put((byte)value);
      break;

    default:
      writeBuffer.put((byte)(value >>> 8));
      writeBuffer.put((byte)value);
      break;
  }
}
 

@SuppressWarnings("unchecked")
@Override
public Object read(Kryo kryo, Input input, Class type) {
    try {
        ObjectMap graphContext = kryo.getGraphContext();
        ObjectInputStream objectStream = (ObjectInputStream) graphContext.get(this);
        if (objectStream == null) {
            objectStream = new ObjectInputStream(input) {
                @Override
                protected Class<?> resolveClass(ObjectStreamClass desc) throws ClassNotFoundException {
                    return ClassUtils.getClass(KryoSerialization.class.getClassLoader(), desc.getName());
                }
            };
            graphContext.put(this, objectStream);
        }
        return objectStream.readObject();
    } catch (Exception ex) {
        throw new KryoException("Error during Java deserialization.", ex);
    }
}
 

/**
 * Tries to copy the given record from using the provided Kryo instance. If this fails, then
 * the record from is copied by serializing it into a byte buffer and deserializing it from
 * there.
 *
 * @param from Element to copy
 * @param reuse Reuse element for the deserialization
 * @param kryo Kryo instance to use
 * @param serializer TypeSerializer which is used in case of a Kryo failure
 * @param <T> Type of the element to be copied
 * @return Copied element
 */
public static <T> T copy(T from, T reuse, Kryo kryo, TypeSerializer<T> serializer) {
	try {
		return kryo.copy(from);
	} catch (KryoException ke) {
		// Kryo could not copy the object --> try to serialize/deserialize the object
		try {
			byte[] byteArray = InstantiationUtil.serializeToByteArray(serializer, from);

			return InstantiationUtil.deserializeFromByteArray(serializer, reuse, byteArray);
		} catch (IOException ioe) {
			throw new RuntimeException("Could not copy object by serializing/deserializing" +
				" it.", ioe);
		}
	}
}
 

/**
 * Deserialize an object.
 * Only supports BIN and BIN_GZ SerializationMode.
 * 
 * If the specified file does not exist or is empty, then this call returns null.
 * 
 * @param inputStream
 * @param type
 * @param mode
 * @return
 * @throws IOException
 * @throws ClassNotFoundException
 */
public static <T> T deserialize(InputStream inStream, Class<T> type, SerializationMode mode) {
  try {
    T object;
    if (mode == SerializationMode.BIN || mode == SerializationMode.BIN_GZ) {
      Kryo kryo = new Kryo();
      kryo.setReferences(false);
      Input input = new Input(mode == SerializationMode.BIN_GZ ? 
          new GZIPInputStream(inStream) : inStream);
      object = kryo.readObject(input, type);
      
    } else {
      throw new UnsupportedOperationException();
    }

    return object;
    
  } catch (KryoException  | IOException e) {
    logger.error("Unable to deserialize {} (mode: {})", mode);
    logger.error("Deserialization exception", e);
    throw new RuntimeException(e);
  }
}
 

@SuppressWarnings({"unchecked", "rawtypes"})
@Override
public T read(Kryo kryo, Input input, Class aClass) {
	try {
		ObjectMap graphContext = kryo.getGraphContext();
		ObjectInputStream objectStream = (ObjectInputStream)graphContext.get(this);
		if (objectStream == null) {
			// make sure we use Kryo's classloader
			objectStream = new InstantiationUtil.ClassLoaderObjectInputStream(input, kryo.getClassLoader());
			graphContext.put(this, objectStream);
		}
		return (T) objectStream.readObject();
	} catch (Exception ex) {
		throw new KryoException("Error during Java deserialization.", ex);
	}
}
 

/**
 * Tries to copy the given record from using the provided Kryo instance. If this fails, then
 * the record from is copied by serializing it into a byte buffer and deserializing it from
 * there.
 *
 * @param from Element to copy
 * @param reuse Reuse element for the deserialization
 * @param kryo Kryo instance to use
 * @param serializer TypeSerializer which is used in case of a Kryo failure
 * @param <T> Type of the element to be copied
 * @return Copied element
 */
public static <T> T copy(T from, T reuse, Kryo kryo, TypeSerializer<T> serializer) {
	try {
		return kryo.copy(from);
	} catch (KryoException ke) {
		// Kryo could not copy the object --> try to serialize/deserialize the object
		try {
			byte[] byteArray = InstantiationUtil.serializeToByteArray(serializer, from);

			return InstantiationUtil.deserializeFromByteArray(serializer, reuse, byteArray);
		} catch (IOException ioe) {
			throw new RuntimeException("Could not copy object by serializing/deserializing" +
				" it.", ioe);
		}
	}
}
 

/**
 * Require makes sure that at least required number of bytes are kept in the buffer. If not, then
 * it will load exactly the difference between required and currently available number of bytes.
 * Thus, it will only load the data which is required and never prefetch data.
 *
 * @param required the number of bytes being available in the buffer
 * @return the number of bytes remaining, which is equal to required
 * @throws KryoException
 */
@Override
protected int require(int required) throws KryoException {
	if(required > capacity) {
		throw new KryoException("Buffer too small: capacity: " + capacity + ", " +
				"required: " + required);
	}

	position = 0;
	int bytesRead = 0;
	int count;
	while(true){
		count = fill(buffer, bytesRead, required - bytesRead);

		if(count == -1){
			throw new KryoException(new EOFException("No more bytes left."));
		}

		bytesRead += count;
		if(bytesRead == required){
			break;
		}
	}
	limit = required;
	return required;
}
 

/**
 * Tries to copy the given record from using the provided Kryo instance. If this fails, then
 * the record from is copied by serializing it into a byte buffer and deserializing it from
 * there.
 *
 * @param from Element to copy
 * @param kryo Kryo instance to use
 * @param serializer TypeSerializer which is used in case of a Kryo failure
 * @param <T> Type of the element to be copied
 * @return Copied element
 */
public static <T> T copy(T from, Kryo kryo, TypeSerializer<T> serializer) {
	try {
		return kryo.copy(from);
	} catch (KryoException ke) {
		// Kryo could not copy the object --> try to serialize/deserialize the object
		try {
			byte[] byteArray = InstantiationUtil.serializeToByteArray(serializer, from);

			return InstantiationUtil.deserializeFromByteArray(serializer, byteArray);
		} catch (IOException ioe) {
			throw new RuntimeException("Could not copy object by serializing/deserializing" +
				" it.", ioe);
		}
	}
}
 

public Registration readClass(Input input) {
	int classID = input.readVarInt(true);
	switch (classID) {
	case Kryo.NULL:
		if (TRACE || (DEBUG && kryo.getDepth() == 1))
			log("Read", null);
		return null;
	case NAME + 2: // Offset for NAME and NULL.
		return readName(input);
	}
	if (classID == memoizedClassId)
		return memoizedClassIdValue;
	Registration registration = idToRegistration.get(classID - 2);
	if (registration == null)
		throw new KryoException("Encountered unregistered class ID: " + (classID - 2));
	if (TRACE)
		trace("kryo", "Read class " + (classID - 2) + ": " + className(registration.getType()));
	memoizedClassId = classID;
	memoizedClassIdValue = registration;
	return registration;
}
 

@SuppressWarnings({"unchecked", "rawtypes"})
@Override
public T read(Kryo kryo, Input input, Class aClass) {
	try {
		ObjectMap graphContext = kryo.getGraphContext();
		ObjectInputStream objectStream = (ObjectInputStream)graphContext.get(this);
		if (objectStream == null) {
			// make sure we use Kryo's classloader
			objectStream = new InstantiationUtil.ClassLoaderObjectInputStream(input, kryo.getClassLoader());
			graphContext.put(this, objectStream);
		}
		return (T) objectStream.readObject();
	} catch (Exception ex) {
		throw new KryoException("Error during Java deserialization.", ex);
	}
}
 

@Override
public void write(int value) throws KryoException
{
  if (!writeBuffer.hasRemaining()) {
    advanceWriteBuffer();
  }
  writeBuffer.put((byte)value);
}
 

@Override
public void write(byte[] bytes) throws KryoException
{
  int remaining = writeBuffer.remaining();
  if (bytes.length > remaining) {
    writeBuffer.put(bytes, 0, remaining);
    advanceWriteBuffer();
    write(bytes, remaining, bytes.length - remaining);
  } else {
    writeBuffer.put(bytes);
  }
}
 

public boolean readBoolean() throws EOFException {
    try {
        return input.readBoolean();
    } catch (KryoException e) {
        throw maybeEndOfStream(e);
    }
}
 

@Override
public void writeByte(int value) throws KryoException
{
  if (!writeBuffer.hasRemaining()) {
    advanceWriteBuffer();
  }
  writeBuffer.put((byte)value);
}
 

public short readShort() throws IOException {
    try {
        return input.readShort();
    } catch (KryoException e) {
        throw new IOException(e);
    }
}
 

public int readInt() throws EOFException {
    try {
        return input.readInt();
    } catch (KryoException e) {
        throw maybeEndOfStream(e);
    }
}
 

public String readUTF() throws IOException {
        // TODO
        try {
//            return kryo.readObject(input, String.class);
            return input.readString();
        } catch (KryoException e) {
            throw new IOException(e);
        }
    }
 

public void readBytes(byte[] buffer, int offset, int count) throws EOFException {
    try {
        input.readBytes(buffer, offset, count);
    } catch (KryoException e) {
        throw maybeEndOfStream(e);
    }
}
 

public int readInt() throws IOException {
    try {
        return input.readInt();
    } catch (KryoException e) {
        throw new IOException(e);
    }
}
 

public String readUTF() throws IOException {
        // TODO
        try {
//            return kryo.readObject(input, String.class);
            return input.readString();
        } catch (KryoException e) {
            throw new IOException(e);
        }
    }
 

public byte readByte() throws EOFException {
    try {
        return input.readByte();
    } catch (KryoException e) {
        throw maybeEndOfStream(e);
    }
}
 

public short readShort() throws IOException {
    try {
        return input.readShort();
    } catch (KryoException e) {
        throw new IOException(e);
    }
}
 

@Override
public void write(int value) throws KryoException
{
  if (!writeBuffer.hasRemaining()) {
    advanceWriteBuffer();
  }
  writeBuffer.put((byte)value);
}
 

public Object readObject() throws IOException, ClassNotFoundException {
        // TODO
//        throw new UnsupportedOperationException();
        try {
            return kryo.readClassAndObject(input);
        } catch (KryoException e) {
            throw new IOException(e);
        }
    }
 

public byte[] readBytes() throws IOException {
    try {
        int len = input.readInt();
        if (len < 0) {
            return null;
        } else if (len == 0) {
            return new byte[]{};
        } else {
            return input.readBytes(len);
        }
    } catch (KryoException e) {
        throw new IOException(e);
    }
}
 

@Override
public boolean readBool() throws IOException {
    try {
        return input.readBoolean();
    } catch (KryoException e) {
        throw new IOException(e);
    }
}
 

public String readNullableString() throws EOFException {
    try {
        return input.readString();
    } catch (KryoException e) {
        throw maybeEndOfStream(e);
    }
}
 

@Override
public float readFloat() throws IOException {
    try {
        return input.readFloat();
    } catch (KryoException e) {
        throw new IOException(e);
    }
}
 

public double readDouble() throws IOException {
    try {
        return input.readDouble();
    } catch (KryoException e) {
        throw new IOException(e);
    }
}
 

public float readFloat() throws IOException {
    try {
        return input.readFloat();
    } catch (KryoException e) {
        throw new IOException(e);
    }
}