The @latticexyz/block-logs-stream package provides utilities for efficiently retrieving blockchain event logs by combining viem for blockchain interaction and RxJS for reactive stream processing.
Key capabilities include:
- Creating streams of block data using
createBlockStream. - Converting block ranges into log streams using
blockRangeToLogs. - Grouping logs by their block number using
groupLogsByBlockNumber.
To use these utilities, you need a viem public client and an ABI parsed via parseAbi.
import { filter, map, mergeMap } from "rxjs";
import { createPublicClient, parseAbi } from "viem";
import { createBlockStream, groupLogsByBlockNumber, blockRangeToLogs } from "@latticexyz/block-logs-stream";
const publicClient = createPublicClient({
// your viem public client config here
});
const latestBlock$ = await createBlockStream({ publicClient, blockTag: "latest" });
const latestBlockNumber$ = latestBlock$.pipe(map((block) => block.number));
latestBlockNumber$
.pipe(
map((latestBlockNumber) => ({ startBlock: 0n, endBlock: latestBlockNumber })),
blockRangeToLogs({
publicClient,
address,
events: parseAbi([
"event Store_SetRecord(bytes32 indexed tableId, bytes32[] keyTuple, bytes staticData, bytes32 encodedLengths, bytes dynamicData)",
"event Store_SpliceStaticData(bytes32 indexed tableId, bytes32[] keyTuple, uint48 start, bytes data)",
"event Store_SpliceDynamicData(bytes32 indexed tableId, bytes32[] keyTuple, uint48 start, uint40 deleteCount, bytes32 encodedLengths, bytes data)",
"event Store_DeleteRecord(bytes32 indexed tableId, bytes32[] keyTuple)",
]),
}),
mergeMap(({ logs }) => from(groupLogsByBlockNumber(logs))),
)
.subscribe((block) => {
console.log("got events for block", block);
});
