Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | 5x 5x 5x 15x 28x 15x 6x 9x 19x 9x 3x 6x 3x 5x 3x 3x 3x 3x 3x 3x 3x 5x 10x 10x 10x 24x 15x 15x 3x 3x 15x 15x 15x 6x 6x 12x 6x 6x 6x 3x 6x 9x 9x 3x 3x 3x 3x 6x 6x 21x 21x 21x 21x 3x 3x | import Debug from "debug";
import EventEmitter from "events";
import {Delta, DeltaID} from "./delta";
import {Lossless} from "./lossless";
import {DomainEntityID, TransactionID} from "./types";
const debug = Debug('rz:transactions');
function getDeltaTransactionId(delta: Delta): TransactionID | undefined {
const {target: transactionId} = delta.pointers.find(({
localContext,
target,
targetContext
}) =>
localContext === "_transaction" &&
typeof target === "string" &&
targetContext === "deltas"
) || {};
if (transactionId && typeof transactionId === "string") {
return transactionId;
}
}
function getTransactionSize(delta: Delta): {
transactionId: TransactionID,
size: number
} | undefined {
const {target: transactionId} = delta.pointers.find(({
localContext,
target,
targetContext
}) =>
localContext === "_transaction" &&
typeof target === "string" &&
targetContext === "size"
) || {};
if (transactionId && typeof transactionId === "string") {
// This delta describes a transaction
const {target: size} = delta.pointers.find(({
localContext,
target
}) =>
localContext === "size" &&
typeof target === "number"
) || {};
return {transactionId, size: size as number};
}
}
export class Transaction {
size?: number;
receivedDeltaIds = new Set<DeltaID>();
entityIds = new Set<DomainEntityID>();
resolved: Promise<boolean>;
constructor(readonly transactions: Transactions, readonly id: TransactionID) {
this.resolved = new Promise((resolve) => {
this.transactions.eventStream.on("completed", (transactionId) => {
if (transactionId === this.id) resolve(true);
});
});
}
getReceivedDeltaIds() {
return Array.from(this.receivedDeltaIds.values());
}
}
export class Transactions {
transactions = new Map<TransactionID, Transaction>();
eventStream = new EventEmitter();
constructor(readonly lossless: Lossless) {}
get(id: TransactionID): Transaction | undefined {
return this.transactions.get(id);
}
getOrInit(id: TransactionID): Transaction {
let t = this.transactions.get(id);
if (!t) {
t = new Transaction(this, id);
this.transactions.set(id, t);
}
return t;
}
ingestDelta(delta: Delta, targets: DomainEntityID[]): TransactionID | undefined {
// This delta may be part of a transaction
{
const transactionId = getDeltaTransactionId(delta);
if (transactionId) {
const t = this.getOrInit(transactionId);
for (const id of targets) {
t.entityIds.add(id);
}
// Add this to the delta's data structure for quick reference
delta.transactionId = transactionId;
// Update our transaction tracking
this.receivedDelta(transactionId, delta.id);
// Notify that the transaction is complete
if (this.isComplete(transactionId)) {
this.eventStream.emit("completed", t.id, t.getReceivedDeltaIds());
}
return transactionId;
}
}
// This delta may describe a transaction
{
const {transactionId, size} = getTransactionSize(delta) || {};
if (transactionId && size) {
debug(`[${this.lossless.rhizomeNode.config.peerId}]`, `Transaction ${transactionId} has size ${size}`);
this.setSize(transactionId, size as number);
// Check if the transaction is complete
Iif (this.isComplete(transactionId)) {
const t = this.getOrInit(transactionId);
this.eventStream.emit("completed", t.id, t.getReceivedDeltaIds());
}
return transactionId;
}
}
}
receivedDelta(id: TransactionID, deltaId: DeltaID) {
const t = this.getOrInit(id);
t.receivedDeltaIds.add(deltaId);
}
isComplete(id: TransactionID) {
const t = this.get(id);
Iif (!t) return false;
Iif (t.size === undefined) return false;
return t.receivedDeltaIds.size === t.size;
}
async waitFor(id: TransactionID) {
const t = this.get(id);
Iif (!t) return;
await t.resolved;
}
setSize(id: TransactionID, size: number) {
const t = this.getOrInit(id);
t.size = size;
}
get ids() {
return Array.from(this.transactions.keys());
}
}
|