package org.apache.cassandra.batchlog;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.UUIDSerializer;
import static org.apache.cassandra.db.TypeSizes.sizeof;
import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
public final class Batch
{
public static final Serializer serializer = new Serializer();
public final UUID id;
public final long creationTime;
final Collection<Mutation> decodedMutations;
final Collection<ByteBuffer> encodedMutations;
private Batch(UUID id, long creationTime, Collection<Mutation> decodedMutations, Collection<ByteBuffer> encodedMutations)
{
this.id = id;
this.creationTime = creationTime;
this.decodedMutations = decodedMutations;
this.encodedMutations = encodedMutations;
}
public static Batch createLocal(UUID id, long creationTime, Collection<Mutation> mutations)
{
return new Batch(id, creationTime, mutations, Collections.emptyList());
}
public static Batch createRemote(UUID id, long creationTime, Collection<ByteBuffer> mutations)
{
return new Batch(id, creationTime, Collections.<Mutation>emptyList(), mutations);
}
public int size()
{
return decodedMutations.size() + encodedMutations.size();
}
static final class Serializer implements IVersionedSerializer<Batch>
{
public long serializedSize(Batch batch, int version)
{
assert batch.encodedMutations.isEmpty() : "attempted to serialize a 'remote' batch";
long size = UUIDSerializer.serializer.serializedSize(batch.id, version);
size += sizeof(batch.creationTime);
size += sizeofUnsignedVInt(batch.decodedMutations.size());
for (Mutation mutation : batch.decodedMutations)
{
int mutationSize = (int) Mutation.serializer.serializedSize(mutation, version);
size += sizeofUnsignedVInt(mutationSize);
size += mutationSize;
}
return size;
}
public void serialize(Batch batch, DataOutputPlus out, int version) throws IOException
{
assert batch.encodedMutations.isEmpty() : "attempted to serialize a 'remote' batch";
UUIDSerializer.serializer.serialize(batch.id, out, version);
out.writeLong(batch.creationTime);
out.writeUnsignedVInt(batch.decodedMutations.size());
for (Mutation mutation : batch.decodedMutations)
{
out.writeUnsignedVInt(Mutation.serializer.serializedSize(mutation, version));
Mutation.serializer.serialize(mutation, out, version);
}
}
public Batch deserialize(DataInputPlus in, int version) throws IOException
{
UUID id = UUIDSerializer.serializer.deserialize(in, version);
long creationTime = in.readLong();
return version == MessagingService.current_version
? createRemote(id, creationTime, readEncodedMutations(in))
: createLocal(id, creationTime, decodeMutations(in, version));
}
private static Collection<ByteBuffer> readEncodedMutations(DataInputPlus in) throws IOException
{
int count = (int) in.readUnsignedVInt();
ArrayList<ByteBuffer> mutations = new ArrayList<>(count);
for (int i = 0; i < count; i++)
mutations.add(ByteBufferUtil.readWithVIntLength(in));
return mutations;
}
private static Collection<Mutation> decodeMutations(DataInputPlus in, int version) throws IOException
{
int count = (int) in.readUnsignedVInt();
ArrayList<Mutation> mutations = new ArrayList<>(count);
for (int i = 0; i < count; i++)
{
in.readUnsignedVInt();
mutations.add(Mutation.serializer.deserialize(in, version));
}
return mutations;
}
}
}