package iotcloud;
import java.util.Set;
+import java.util.Map;
import java.util.HashSet;
import javax.script.ScriptException;
import java.lang.NullPointerException;
+import java.nio.ByteBuffer;
class PendingTransaction {
- static final byte Equal = Guard.Equal;
- static final byte NotEqual = Guard.NotEqual;
-
- private Set<KeyValue> keyValueUpdateSet;
- private Guard guard;
+ private Set<KeyValue> keyValueUpdateSet = null;
+ private Set<KeyValue> keyValueGuardSet = null;
private long arbitrator = -1;
+ private long clientLocalSequenceNumber = -1;
+ private long machineId = -1;
+
+ private int currentDataSize = 0;
- public PendingTransaction() {
+ public PendingTransaction(long _machineId) {
+ machineId = _machineId;
keyValueUpdateSet = new HashSet<KeyValue>();
- guard = new Guard();
+ keyValueGuardSet = new HashSet<KeyValue>();
}
/**
// Remove key if we are adding a newer version of the same key
if (rmKV != null) {
keyValueUpdateSet.remove(rmKV);
+ currentDataSize -= rmKV.getSize();
}
// Add the key to the hash set
keyValueUpdateSet.add(newKV);
+ currentDataSize += newKV.getSize();
}
+ /**
+ * Add a new key value to the guard set
+ *
+ */
+ public void addKVGuard(KeyValue newKV) {
+ // Add the key to the hash set
+ keyValueGuardSet.add(newKV);
+ currentDataSize += newKV.getSize();
+ }
+
+ /**
+ * Checks if the arbitrator is the same
+ */
public boolean checkArbitrator(long arb) {
if (arbitrator == -1) {
arbitrator = arb;
return arb == arbitrator;
}
-
+ /**
+ * Get the transaction arbitrator
+ */
+ public long getArbitrator() {
+ return arbitrator;
+ }
/**
* Get the key value update set
- *
*/
public Set<KeyValue> getKVUpdates() {
return keyValueUpdateSet;
}
/**
- * Get the guard
- *
- */
- public Guard getGuard() {
- return guard;
+ * Get the key value update set
+ */
+ public Set<KeyValue> getKVGuard() {
+ return keyValueGuardSet;
}
- /**
- * Add a guard to this transaction
- *
- */
- public void addGuard(Guard _guard) {
- guard = _guard;
+ public void setClientLocalSequenceNumber(long _clientLocalSequenceNumber) {
+ clientLocalSequenceNumber = _clientLocalSequenceNumber;
}
- /**
- * Evaluate the guard expression for a given transaction using a set of key value pairs.
- *
- */
- public boolean evaluate(Set<KeyValue> kvSet) throws ScriptException, NullPointerException {
+ public long getClientLocalSequenceNumber() {
+ return clientLocalSequenceNumber;
+ }
- // Evaluate the guard using the current KV Set
- return guard.evaluate(kvSet);
+ public long getMachineId() {
+ return machineId;
}
- /**
- * Add a boolean expression to the guard.
- *
- */
- public void setGuardExpression(String expr) {
- guard.setGuardExpression(expr);
+ public boolean evaluateGuard(Map<IoTString, KeyValue> keyValTableCommitted, Map<IoTString, KeyValue> keyValTableSpeculative, Map<IoTString, KeyValue> keyValTablePendingTransSpeculative) {
+ for (KeyValue kvGuard : keyValueGuardSet) {
+
+ // First check if the key is in the speculative table, this is the value of the latest assumption
+ KeyValue kv = keyValTablePendingTransSpeculative.get(kvGuard.getKey());
+
+
+ if (kv == null) {
+ // if it is not in the pending trans table then check the speculative table and use that
+ // value as our latest assumption
+ kv = keyValTableSpeculative.get(kvGuard.getKey());
+ }
+
+
+ if (kv == null) {
+ // if it is not in the speculative table then check the committed table and use that
+ // value as our latest assumption
+ kv = keyValTableCommitted.get(kvGuard.getKey());
+ }
+
+ if (kvGuard.getValue() != null) {
+ if ((kv == null) || (!kvGuard.getValue().equals(kv.getValue()))) {
+ return false;
+ }
+ } else {
+ if (kv != null) {
+ return false;
+ }
+ }
+ }
+ return true;
}
- /**
- * Trampoline static method.
- *
- */
- public static String createExpression(IoTString keyName, IoTString keyValue, byte op) {
- return Guard.createExpression(keyName, keyValue, op);
+ public Transaction createTransaction() {
+
+ Transaction newTransaction = new Transaction();
+ int transactionPartCount = 0;
+
+ // Convert all the data into a byte array so we can start partitioning
+ byte[] byteData = convertDataToBytes();
+
+ int currentPosition = 0;
+ int remaining = byteData.length;
+
+ while (remaining > 0) {
+
+ Boolean isLastPart = false;
+ // determine how much to copy
+ int copySize = TransactionPart.MAX_NON_HEADER_SIZE;
+ if (remaining <= TransactionPart.MAX_NON_HEADER_SIZE) {
+ copySize = remaining;
+ isLastPart = true; // last bit of data so last part
+ }
+
+ // Copy to a smaller version
+ byte[] partData = new byte[copySize];
+ System.arraycopy(byteData, currentPosition, partData, 0, copySize);
+
+ TransactionPart part = new TransactionPart(null, machineId, arbitrator, clientLocalSequenceNumber, transactionPartCount, partData, isLastPart);
+ newTransaction.addPartEncode(part);
+
+ // Update position, count and remaining
+ currentPosition += copySize;
+ transactionPartCount++;
+ remaining -= copySize;
+ }
+
+ // Add the Guard Conditions
+ for (KeyValue kv : keyValueGuardSet) {
+ newTransaction.addGuardKV(kv);
+ }
+
+ // Add the updates
+ for (KeyValue kv : keyValueUpdateSet) {
+ newTransaction.addUpdateKV(kv);
+ }
+
+ return newTransaction;
+ }
+
+ private byte[] convertDataToBytes() {
+
+ // Calculate the size of the data
+ int sizeOfData = 2 * Integer.BYTES; // Number of Update KV's and Guard KV's
+ sizeOfData += currentDataSize;
+
+ // Data handlers and storage
+ byte[] dataArray = new byte[sizeOfData];
+ ByteBuffer bbEncode = ByteBuffer.wrap(dataArray);
+
+ // Encode the size of the updates and guard sets
+ bbEncode.putInt(keyValueGuardSet.size());
+ bbEncode.putInt(keyValueUpdateSet.size());
+
+ // Encode all the guard conditions
+ for (KeyValue kv : keyValueGuardSet) {
+ kv.encode(bbEncode);
+ }
+
+ // Encode all the updates
+ for (KeyValue kv : keyValueUpdateSet) {
+ kv.encode(bbEncode);
+ }
+
+ return bbEncode.array();
}
}
\ No newline at end of file
projects / iotcloud.git / blobdiff