Changes logs added/removed callbacks to receive an array of all logs in a particular block.

Also fires callback with empty array when block contains no logs matching filters.

Also includes blockHash with callback.

This change makes it easier for consumers to know when they are done receiving logs for a particular block.  With this knowledge, they can now process block + logs as a single atomic operation if they desire by saving off the block and waiting for a matching set of logs to come through (or notification of block removal in case log fetching fails).

This change somewhat constrains the problem being solved by this library to just "getting ordering right", rather than trying to provide a nice interface into the stream.  It isn't difficult for users of this library to split apart the logs into multiple callbacks with a _very_ tiny callback wrapper, yet this change opens the doors to a number of new use cases.

Author: MicahZoltu

package.json

@@ -1,6 +1,6 @@
 {
   "name": "ethereumjs-blockstream",
-  "version": "6.0.1",
+  "version": "7.0.0",
   "description": "A library to turn an unreliable remote source of Ethereum blocks into a reliable stream of blocks with removals on re-orgs and backfills on skips.",
   "main": "output/source/index.js",
   "types": "output/source/index.d.ts",

source/block-and-log-streamer.ts

@@ -30,8 +30,8 @@ export class BlockAndLogStreamer<TBlock extends Block, TLog extends Log> {
 	private readonly logFilters: { [propName: string]: Filter } = {}
 	private readonly onBlockAddedSubscribers: { [propName: string]: (block: TBlock) => void } = {};
 	private readonly onBlockRemovedSubscribers: { [propName: string]: (block: TBlock) => void } = {};
-	private readonly onLogAddedSubscribers: { [propName: string]: (log: TLog) => void } = {};
-	private readonly onLogRemovedSubscribers: { [propName: string]: (log: TLog) => void } = {};
+	private readonly onLogsAddedSubscribers: { [propName: string]: (blockHash: string, logs: Array<TLog>) => void } = {};
+	private readonly onLogsRemovedSubscribers: { [propName: string]: (blockHash: string, logs: Array<TLog>) => void } = {};
 
 	/**
 	 * @param getBlockByHash async function that returns a block given a particular hash or null/throws if the block is not found
@@ -81,12 +81,12 @@ export class BlockAndLogStreamer<TBlock extends Block, TLog extends Log> {
 			.forEach(callback => this.pendingCallbacks.push(() => callback(block)));
 
 		const logFilters = Object.keys(this.logFilters).map(key => this.logFilters[key]);
-		this.logHistory = reconcileLogHistoryWithAddedBlock(this.getLogs, this.logHistory, block, this.onLogAdded, logFilters, this.blockRetention);
+		this.logHistory = reconcileLogHistoryWithAddedBlock(this.getLogs, this.logHistory, block, this.onLogsAdded, logFilters, this.blockRetention);
 		await this.logHistory;
 	};
 
 	private readonly onBlockRemoved = async (block: TBlock): Promise<void> => {
-		this.logHistory = reconcileLogHistoryWithRemovedBlock(this.logHistory, block, this.onLogRemoved);
+		this.logHistory = reconcileLogHistoryWithRemovedBlock(this.logHistory, block, this.onLogsRemoved);
 		await this.logHistory;
 
 		Object.keys(this.onBlockRemovedSubscribers)
@@ -95,18 +95,18 @@ export class BlockAndLogStreamer<TBlock extends Block, TLog extends Log> {
 			.forEach(callback => this.pendingCallbacks.push(() => callback(block)));
 	};
 
-	private readonly onLogAdded = async (log: TLog): Promise<void> => {
-		Object.keys(this.onLogAddedSubscribers)
-			.map((key: string) => this.onLogAddedSubscribers[key])
+	private readonly onLogsAdded = async (blockHash: string, logs: Array<TLog>): Promise<void> => {
+		Object.keys(this.onLogsAddedSubscribers)
+			.map((key: string) => this.onLogsAddedSubscribers[key])
 			.map(callback => logAndSwallowWrapper(callback, this.onError))
-			.forEach(callback => this.pendingCallbacks.push(() => callback(log)));
+			.forEach(callback => this.pendingCallbacks.push(() => callback(blockHash, logs)));
 	};
 
-	private readonly onLogRemoved = async (log: TLog): Promise<void> => {
-		Object.keys(this.onLogRemovedSubscribers)
-			.map((key: string) => this.onLogRemovedSubscribers[key])
+	private readonly onLogsRemoved = async (blockHash: string, logs: Array<TLog>): Promise<void> => {
+		Object.keys(this.onLogsRemovedSubscribers)
+			.map((key: string) => this.onLogsRemovedSubscribers[key])
 			.map(callback => logAndSwallowWrapper(callback, this.onError))
-			.forEach(callback => this.pendingCallbacks.push(() => callback(log)));
+			.forEach(callback => this.pendingCallbacks.push(() => callback(blockHash, logs)));
 	};
 
 
@@ -151,34 +151,34 @@ export class BlockAndLogStreamer<TBlock extends Block, TLog extends Log> {
 	};
 
 
-	public readonly subscribeToOnLogAdded = (onLogAdded: (log: TLog) => void): string => {
+	public readonly subscribeToOnLogsAdded = (onLogsAdded: (blockHash: string, logs: Array<TLog>) => void): string => {
 		const uuid = `on log added token ${createUuid()}`;
-		this.onLogAddedSubscribers[uuid] = onLogAdded;
+		this.onLogsAddedSubscribers[uuid] = onLogsAdded;
 		return uuid;
 	};
 
-	public readonly unsubscribeFromOnLogAdded = (token: string) => {
+	public readonly unsubscribeFromOnLogsAdded = (token: string) => {
 		if (!token.startsWith("on log added token ")) throw new Error(`Expected a log added subscription token.  Actual: ${token}`);
-		delete this.onLogAddedSubscribers[token];
+		delete this.onLogsAddedSubscribers[token];
 	};
 
 
-	public readonly subscribeToOnLogRemoved = (onLogRemoved: (log: TLog) => void): string => {
+	public readonly subscribeToOnLogsRemoved = (onLogsRemoved: (blockHash: string, logs: Array<TLog>) => void): string => {
 		const uuid = `on log removed token ${createUuid()}`;
-		this.onLogRemovedSubscribers[uuid] = onLogRemoved;
+		this.onLogsRemovedSubscribers[uuid] = onLogsRemoved;
 		return uuid;
 	};
 
-	public readonly unsubscribeFromOnLogRemoved = (token: string) => {
+	public readonly unsubscribeFromOnLogsRemoved = (token: string) => {
 		if (!token.startsWith("on log removed token ")) throw new Error(`Expected a log added subscription token.  Actual: ${token}`);
-		delete this.onLogRemovedSubscribers[token];
+		delete this.onLogsRemovedSubscribers[token];
 	};
 }
 
-function logAndSwallowWrapper<T>(callback: (arg: T) => void, onError: (error: Error) => void): (arg: T) => void {
-	return function (parameter) {
+function logAndSwallowWrapper<T, U>(callback: (arg1?: T, arg2?: U) => void, onError: (error: Error) => void): (arg1?: T, arg2?: U) => void {
+	return function (parameter1, parameter2) {
 		try {
-			callback(parameter);
+			callback(parameter1, parameter2);
 		} catch (error) {
 			onError(error);
 		}

source/log-reconciler.ts

@@ -8,13 +8,13 @@ export const reconcileLogHistoryWithAddedBlock = async <TBlock extends Block, TL
 	getLogs: (filterOptions: FilterOptions) => Promise<TLog[]>,
 	logHistory: LogHistory<TLog> | Promise<LogHistory<TLog>>,
 	newBlock: TBlock,
-	onLogAdded: (log: TLog) => Promise<void>,
+	onLogsAdded: (blockHash: string, logs: Array<TLog>) => Promise<void>,
 	filters: Filter[] = [],
 	historyBlockLength: number = 100,
 ): Promise<LogHistory<TLog>> => {
 	logHistory = await logHistory;
 	const logs = await getFilteredLogs(getLogs, newBlock, filters);
-	logHistory = await addNewLogsToHead(logHistory, logs, onLogAdded);
+	logHistory = await addNewLogsToHead(newBlock.hash, logHistory, logs, onLogsAdded);
 	logHistory = await pruneOldLogs(logHistory, newBlock, historyBlockLength);
 	return logHistory;
 }
@@ -27,14 +27,18 @@ const getFilteredLogs = async <TBlock extends Block, TLog extends Log>(getLogs:
 	return nestedLogs.reduce((allLogs, logs) => allLogs.concat(logs), []);
 }
 
-const addNewLogsToHead = async <TLog extends Log>(logHistory: LogHistory<TLog>, newLogs: Array<TLog>, onLogAdded: (log: TLog) => Promise<void>): Promise<LogHistory<TLog>> => {
+const addNewLogsToHead = async <TLog extends Log>(blockHash: string, logHistory: LogHistory<TLog>, newLogs: Array<TLog>, onLogsAdded: (blockHash: string, logs: Array<TLog>) => Promise<void>): Promise<LogHistory<TLog>> => {
 	const sortedLogs = newLogs.sort((logA, logB) => parseHexInt(logA.logIndex) - parseHexInt(logB.logIndex));
+	const addedLogs: Array<TLog> = []
 	for (const logToAdd of sortedLogs) {
 		// we may already have this log because two filters can return the same log
 		if (logHistory.some(logInHistory => logInHistory!.blockHash === logToAdd.blockHash && logInHistory!.logIndex === logToAdd.logIndex)) continue;
 		ensureOrder(logHistory.last(), logToAdd);
-		logHistory = await addNewLogToHead(logHistory, logToAdd, onLogAdded);
+		logHistory = logHistory.push(logToAdd)
+		addedLogs.push(logToAdd)
 	}
+	// CONSIDER: the user getting this notification won't have any visibility into the updated log history yet. should we announce new logs in a `setTimeout`? should we provide log history with new logs?
+	await onLogsAdded(blockHash, addedLogs)
 	return logHistory;
 }
 
@@ -43,13 +47,6 @@ const pruneOldLogs = async <TBlock extends Block, TLog extends Log>(logHistory:
 	return logHistory.skipUntil(log => parseHexInt(newBlock.number) - parseHexInt(log!.blockNumber) < historyBlockLength).toList();
 }
 
-const addNewLogToHead = async <TLog extends Log>(logHistory: LogHistory<TLog>, newLog: TLog, onLogAdded: (log: TLog) => Promise<void>): Promise<LogHistory<TLog>> => {
-	logHistory = logHistory.push(newLog);
-	// CONSIDER: the user getting this notification won't have any visibility into the updated log history yet. should we announce new logs in a `setTimeout`? should we provide log history with new logs?
-	await onLogAdded(newLog);
-	return logHistory;
-}
-
 const ensureOrder = <TLog extends Log>(headLog: TLog | undefined, newLog: TLog) => {
 	if (headLog === undefined) return;
 	const headBlockNumber = parseHexInt(headLog.blockNumber);
@@ -64,14 +61,16 @@ const ensureOrder = <TLog extends Log>(headLog: TLog | undefined, newLog: TLog)
 export const reconcileLogHistoryWithRemovedBlock = async <TBlock extends Block, TLog extends Log>(
 	logHistory: LogHistory<TLog>|Promise<LogHistory<TLog>>,
 	removedBlock: TBlock,
-	onLogRemoved: (log: TLog) => Promise<void>,
+	onLogsRemoved: (blockHash: string, logs: Array<TLog>) => Promise<void>,
 ): Promise<LogHistory<TLog>> => {
 	logHistory = await logHistory;
 
+	const removedLogs = []
 	while (!logHistory.isEmpty() && logHistory.last().blockHash === removedBlock.hash) {
-		await onLogRemoved(logHistory.last());
+		removedLogs.push(logHistory.last());
 		logHistory = logHistory.pop();
 	}
+	await onLogsRemoved(removedBlock.hash, removedLogs);
 
 	// sanity check, no known way to trigger the error
 	if (logHistory.some(log => log!.blockHash === removedBlock.hash)) throw new Error("found logs for removed block not at head of log history");