Adds support for a wider variety of Block and Log models.

Depending on the library you are using to provide blocks/logs to blockstream, your Block and Log model may not align exactly with the model this library was using.  Now as long as you meet some very basic minimum interface requirements you can use any Block/Log model you want.

Author: MicahZoltu

.travis.yml

@@ -3,8 +3,7 @@ sudo: false
 language: node_js
 
 node_js:
-  - "7"
-  - "6"
+  - "8"
 
 before_script:
   - npm install

package-lock.json

@@ -35,7 +35,7 @@
     "typescript": "2.2.2"
   },
   "dependencies": {
-    "immutable": "3.8.1",
+    "immutable": "3.8.2",
     "source-map-support": "0.4.14",
     "uuid": "3.0.1"
   }

package.json

@@ -9,66 +9,68 @@ import { reconcileLogHistoryWithAddedBlock, reconcileLogHistoryWithRemovedBlock
 import { List as ImmutableList } from "immutable";
 import * as createUuid from "uuid";
 
-export class BlockAndLogStreamer {
-	private blockHistory: Promise<BlockHistory> = Promise.resolve(ImmutableList<Block>());
-	private logHistory: Promise<LogHistory> = Promise.resolve(ImmutableList<Log>());
-	private latestBlock: Block | null = null;
+export class BlockAndLogStreamer<TBlock extends Block, TLog extends Log> {
+	private blockHistory: Promise<BlockHistory<TBlock>> = Promise.resolve(ImmutableList<TBlock>());
+	private logHistory: Promise<LogHistory<TLog>> = Promise.resolve(ImmutableList<TLog>());
+	private latestBlock: TBlock | null = null;
 
 	private readonly blockRetention: number;
 
-	private readonly getBlockByHash: (hash: string) => Promise<Block | null>;
-	private readonly getLogs: (filterOptions: FilterOptions) => Promise<Log[]>;
+	private readonly getBlockByHash: (hash: string) => Promise<TBlock | null>;
+	private readonly getLogs: (filterOptions: FilterOptions) => Promise<TLog[]>;
 
 	private readonly logFilters: { [propName: string]: Filter } = {}
-	private readonly onBlockAddedSubscribers: { [propName: string]: (block: Block) => void } = {};
-	private readonly onBlockRemovedSubscribers: { [propName: string]: (block: Block) => void } = {};
-	private readonly onLogAddedSubscribers: { [propName: string]: (log: Log) => void } = {};
-	private readonly onLogRemovedSubscribers: { [propName: string]: (log: Log) => void } = {};
+	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 } = {};
 
 	constructor(
-		getBlockByHash: (hash: string) => Promise<Block | null>,
-		getLogs: (filterOptions: FilterOptions) => Promise<Log[]>,
+		getBlockByHash: (hash: string) => Promise<TBlock | null>,
+		getLogs: (filterOptions: FilterOptions) => Promise<TLog[]>,
 		configuration?: { blockRetention?: number },
 	) {
 		this.getBlockByHash = getBlockByHash;
 		this.getLogs = getLogs;
 		this.blockRetention = (configuration && configuration.blockRetention) ? configuration.blockRetention : 100;
 	}
 
-	static createCallbackStyle = (
-		getBlockByHash: (hash: string, callback: (error?: Error, block?: Block | null) => void) => void,
-		getLogs: (filterOptions: FilterOptions, callback: (error?: Error, logs?: Log[]) => void) => void,
+	static createCallbackStyle = <TBlock extends Block, TLog extends Log>(
+		getBlockByHash: (hash: string, callback: (error?: Error, block?: TBlock | null) => void) => void,
+		getLogs: (filterOptions: FilterOptions, callback: (error?: Error, logs?: TLog[]) => void) => void,
 		configuration?: { blockRetention?: number },
-	): BlockAndLogStreamer => {
-		const wrappedGetBlockByHash = (hash: string): Promise<Block | null> => new Promise<Block | null>((resolve, reject) => {
-			getBlockByHash(hash, (error, block) => {
-				if (error) throw error;
-				else resolve(block);
+	): BlockAndLogStreamer<TBlock, TLog> => {
+		const wrappedGetBlockByHash = (hash: string): Promise<TBlock | null> => {
+			return new Promise<TBlock | null>((resolve, reject) => {
+				getBlockByHash(hash, (error, block) => {
+					if (error) throw error;
+					else resolve(block);
+				});
 			});
-		});
-		const wrappedGetLogs = (filterOptions: FilterOptions): Promise<Log[]> => new Promise<Log[]>((resolve, reject) => {
+		};
+		const wrappedGetLogs = (filterOptions: FilterOptions): Promise<Array<TLog>> => new Promise<Array<TLog>>((resolve, reject) => {
 			getLogs(filterOptions, (error, logs) => {
 				if (error) throw error;
 				if (!logs) throw new Error("Received null/undefined logs and no error.");
 				resolve(logs);
 			});
 		});
-		return new BlockAndLogStreamer(wrappedGetBlockByHash, wrappedGetLogs, configuration);
+		return new BlockAndLogStreamer<TBlock, TLog>(wrappedGetBlockByHash, wrappedGetLogs, configuration);
 	}
 
-	public readonly reconcileNewBlock = async (block: Block): Promise<void> => {
+	public readonly reconcileNewBlock = async (block: TBlock): Promise<void> => {
 		this.blockHistory = reconcileBlockHistory(this.getBlockByHash, this.blockHistory, block, this.onBlockAdded, this.onBlockRemoved, this.blockRetention);
 		const blockHistory = await this.blockHistory;
 		this.latestBlock = blockHistory.last();
 	};
 
-	public readonly reconcileNewBlockCallbackStyle = async (block: Block, callback: (error?: Error) => void): Promise<void> => {
+	public readonly reconcileNewBlockCallbackStyle = async (block: TBlock, callback: (error?: Error) => void): Promise<void> => {
 		this.reconcileNewBlock(block)
 			.then(() => callback(undefined))
 			.catch(error => callback(error));
 	};
 
-	private readonly onBlockAdded = async (block: Block): Promise<void> => {
+	private readonly onBlockAdded = async (block: TBlock): Promise<void> => {
 		const logFilters = Object.keys(this.logFilters).map(key => this.logFilters[key]);
 		this.logHistory = reconcileLogHistoryWithAddedBlock(this.getLogs, this.logHistory, block, this.onLogAdded, logFilters, this.blockRetention);
 
@@ -79,7 +81,7 @@ export class BlockAndLogStreamer {
 			.forEach(callback => callback(block));
 	};
 
-	private readonly onBlockRemoved = async (block: Block): Promise<void> => {
+	private readonly onBlockRemoved = async (block: TBlock): Promise<void> => {
 		this.logHistory = reconcileLogHistoryWithRemovedBlock(this.logHistory, block, this.onLogRemoved);
 
 		await this.logHistory;
@@ -89,22 +91,22 @@ export class BlockAndLogStreamer {
 			.forEach(callback => callback(block));
 	};
 
-	private readonly onLogAdded = async (log: Log): Promise<void> => {
+	private readonly onLogAdded = async (log: TLog): Promise<void> => {
 		Object.keys(this.onLogAddedSubscribers)
 			.map((key: string) => this.onLogAddedSubscribers[key])
 			.map(callback => logAndSwallowWrapper(callback))
 			.forEach(callback => callback(log));
 	};
 
-	private readonly onLogRemoved = async (log: Log): Promise<void> => {
+	private readonly onLogRemoved = async (log: TLog): Promise<void> => {
 		Object.keys(this.onLogRemovedSubscribers)
 			.map((key: string) => this.onLogRemovedSubscribers[key])
 			.map(callback => logAndSwallowWrapper(callback))
 			.forEach(callback => callback(log));
 	};
 
 
-	public readonly getLatestReconciledBlock = (): Block | null => {
+	public readonly getLatestReconciledBlock = (): TBlock | null => {
 		return this.latestBlock;
 	};
 
@@ -121,7 +123,7 @@ export class BlockAndLogStreamer {
 	};
 
 
-	public readonly subscribeToOnBlockAdded = (onBlockAdded: (block: Block) => void): string => {
+	public readonly subscribeToOnBlockAdded = (onBlockAdded: (block: TBlock) => void): string => {
 		const uuid = `on block added token ${createUuid()}`;
 		this.onBlockAddedSubscribers[uuid] = onBlockAdded;
 		return uuid;
@@ -133,7 +135,7 @@ export class BlockAndLogStreamer {
 	};
 
 
-	public readonly subscribeToOnBlockRemoved = (onBlockRemoved: (block: Block) => void): string => {
+	public readonly subscribeToOnBlockRemoved = (onBlockRemoved: (block: TBlock) => void): string => {
 		const uuid = `on block removed token ${createUuid()}`;
 		this.onBlockRemovedSubscribers[uuid] = onBlockRemoved;
 		return uuid;
@@ -145,7 +147,7 @@ export class BlockAndLogStreamer {
 	};
 
 
-	public readonly subscribeToOnLogAdded = (onLogAdded: (log: Log) => void): string => {
+	public readonly subscribeToOnLogAdded = (onLogAdded: (log: TLog) => void): string => {
 		const uuid = `on log added token ${createUuid()}`;
 		this.onLogAddedSubscribers[uuid] = onLogAdded;
 		return uuid;
@@ -157,7 +159,7 @@ export class BlockAndLogStreamer {
 	};
 
 
-	public readonly subscribeToOnLogRemoved = (onLogRemoved: (log: Log) => void): string => {
+	public readonly subscribeToOnLogRemoved = (onLogRemoved: (log: TLog) => void): string => {
 		const uuid = `on log removed token ${createUuid()}`;
 		this.onLogRemovedSubscribers[uuid] = onLogRemoved;
 		return uuid;

source/block-and-log-streamer.ts

@@ -2,14 +2,16 @@ import { Block } from "./models/block";
 import { BlockHistory } from "./models/block-history";
 import { List as ImmutableList } from "immutable";
 
-export const reconcileBlockHistory = async (
-	getBlockByHash: (hash: string) => Promise<Block|null>,
-	blockHistory: BlockHistory|Promise<BlockHistory>,
-	newBlock: Block,
-	onBlockAdded: (block: Block) => Promise<void>,
-	onBlockRemoved: (block: Block) => Promise<void>,
+type GetBlockByHash<TBlock> = (hash: string) => Promise<TBlock|null>;
+
+export const reconcileBlockHistory = async <TBlock extends Block>(
+	getBlockByHash: GetBlockByHash<TBlock>,
+	blockHistory: BlockHistory<TBlock>|Promise<BlockHistory<TBlock>>,
+	newBlock: TBlock,
+	onBlockAdded: (block: TBlock) => Promise<void>,
+	onBlockRemoved: (block: TBlock) => Promise<void>,
 	blockRetention: number = 100,
-): Promise<BlockHistory> => {
+): Promise<BlockHistory<TBlock>> => {
 	blockHistory = await blockHistory;
 	if (isFirstBlock(blockHistory))
 		return await addNewHeadBlock(blockHistory, newBlock, onBlockAdded, blockRetention);
@@ -30,15 +32,15 @@ export const reconcileBlockHistory = async (
 	return await backfill(getBlockByHash, blockHistory, newBlock, onBlockAdded, onBlockRemoved, blockRetention);
 }
 
-const rollback = async (blockHistory: BlockHistory, onBlockRemoved: (block: Block) => Promise<void>): Promise<BlockHistory> => {
+const rollback = async <TBlock extends Block>(blockHistory: BlockHistory<TBlock>, onBlockRemoved: (block: TBlock) => Promise<void>): Promise<BlockHistory<TBlock>> => {
 	while (!blockHistory.isEmpty()) {
 		// CONSIDER: if this throws an exception, removals may have been announced that are actually still in history since throwing will result in no history update. we can't catch errors here because there isn't a clear way to recover from them, the failure may be a downstream system telling us that the block removal isn't possible because they are in a bad state. we could try re-announcing the successfully added blocks, but there would still be a problem with the failed block (should it be re-announced?) and the addition announcements may also fail
 		blockHistory = await removeHeadBlock(blockHistory, onBlockRemoved);
 	}
 	return blockHistory;
 }
 
-const backfill = async (getBlockByHash: (hash: string) => Promise<Block|null>, blockHistory: BlockHistory, newBlock: Block, onBlockAdded: (block: Block) => Promise<void>, onBlockRemoved: (block: Block) => Promise<void>, blockRetention: number) => {
+const backfill = async <TBlock extends Block>(getBlockByHash: GetBlockByHash<TBlock>, blockHistory: BlockHistory<TBlock>, newBlock: TBlock, onBlockAdded: (block: TBlock) => Promise<void>, onBlockRemoved: (block: TBlock) => Promise<void>, blockRetention: number) => {
 	if (newBlock.parentHash === "0x0000000000000000000000000000000000000000000000000000000000000000")
 		return rollback(blockHistory, onBlockRemoved);
 	const parentBlock = await getBlockByHash(newBlock.parentHash);
@@ -49,36 +51,36 @@ const backfill = async (getBlockByHash: (hash: string) => Promise<Block|null>, b
 	return await reconcileBlockHistory(getBlockByHash, blockHistory, newBlock, onBlockAdded, onBlockRemoved, blockRetention);
 }
 
-const addNewHeadBlock = async (blockHistory: BlockHistory, newBlock: Block, onBlockAdded: (block: Block) => Promise<void>, blockRetention: number): Promise<BlockHistory> => {
+const addNewHeadBlock = async <TBlock extends Block>(blockHistory: BlockHistory<TBlock>, newBlock: TBlock, onBlockAdded: (block: TBlock) => Promise<void>, blockRetention: number): Promise<BlockHistory<TBlock>> => {
 	// this is here as a final sanity check, in case we somehow got into an unexpected state, there are no known (and should never be) ways to reach this exception
 	if (!blockHistory.isEmpty() && blockHistory.last().hash !== newBlock.parentHash) throw new Error("New head block's parent isn't our current head.");
-	// CONSIDER: the user getting this notification won't have any visibility into the updated block history yet. should we announce new blocks in a `setTimeout`? should we provide block history with new logs? an announcement failure will result in unwinding the stack and returning the original blockHistory, if we are in the process of backfilling we may have already announced previous blocks that won't actually end up in history (they won't get removed if a re-org occurs and may be re-announced). we can't catch errors thrown by the callback be cause it may be trying to signal to use that the block has become invalid and is un-processable
+	// CONSIDER: the user getting this notification won't have any visibility into the updated block history yet. should we announce new blocks in a `setTimeout`? should we provide block history with new logs? an announcement failure will result in unwinding the stack and returning the original blockHistory, if we are in the process of backfilling we may have already announced previous blocks that won't actually end up in history (they won't get removed if a re-org occurs and may be re-announced). we can't catch errors thrown by the callback because it may be trying to signal to use that the block has become invalid and is un-processable
 	await onBlockAdded(newBlock);
 	blockHistory = blockHistory.push(newBlock);
 	return blockHistory.takeLast(blockRetention).toList();
 }
 
-const removeHeadBlock = async (blockHistory: BlockHistory, onBlockRemoved: (block: Block) => Promise<void>): Promise<BlockHistory> => {
+const removeHeadBlock = async <TBlock extends Block>(blockHistory: BlockHistory<TBlock>, onBlockRemoved: (block: TBlock) => Promise<void>): Promise<BlockHistory<TBlock>> => {
 	let removedBlock = blockHistory.last();
 	blockHistory = blockHistory.pop();
 	await onBlockRemoved(removedBlock);
 	return blockHistory;
 }
 
-const isFirstBlock = (blockHistory: BlockHistory, ): boolean => {
+const isFirstBlock = <TBlock extends Block>(blockHistory: BlockHistory<TBlock>, ): boolean => {
 	return blockHistory.isEmpty();
 }
 
-const isAlreadyInHistory = (blockHistory: BlockHistory, newBlock: Block): boolean => {
+const isAlreadyInHistory = <TBlock extends Block>(blockHistory: BlockHistory<TBlock>, newBlock: TBlock): boolean => {
 	// `block!` is required until the next version of `immutable` is published to NPM (current version 3.8.1) which improves the type definitions
 	return blockHistory.some(block => block!.hash === newBlock.hash);
 }
 
-const isNewHeadBlock = (blockHistory: BlockHistory, newBlock: Block): boolean => {
+const isNewHeadBlock = <TBlock extends Block>(blockHistory: BlockHistory<TBlock>, newBlock: TBlock): boolean => {
 	return blockHistory.last().hash === newBlock.parentHash;
 }
 
-const parentHashIsInHistory = (blockHistory: BlockHistory, newBlock: Block): boolean => {
+const parentHashIsInHistory = <TBlock extends Block>(blockHistory: BlockHistory<TBlock>, newBlock: TBlock): boolean => {
 	// `block!` is required until the next version of `immutable` is published to NPM (current version 3.8.1) which improves the type definitions
 	return blockHistory.some(block => block!.hash === newBlock.parentHash);
 }