Merge pull request #436 from Tenderly/filters-subscriptions

Filters subscriptions

packages/buidler-core/src/internal/buidler-evm/provider/filter.ts

@@ -0,0 +1,142 @@
+import Bloom from "@nomiclabs/ethereumjs-vm/dist/bloom";
+import { BN, bufferToHex, toBuffer } from "ethereumjs-util";
+
+import { RpcLogOutput } from "./output";
+
+export const LATEST_BLOCK = new BN(-1);
+
+export enum Type {
+  LOGS_SUBSCRIPTION = 0,
+  PENDING_TRANSACTION_SUBSCRIPTION = 1,
+  BLOCK_SUBSCRIPTION = 2
+}
+
+export interface FilterCriteria {
+  fromBlock: BN;
+  toBlock: BN;
+  addresses: Buffer[];
+  normalizedTopics: Array<Array<Buffer | null> | null>;
+}
+
+export interface Filter {
+  id: string;
+  type: Type;
+  criteria?: FilterCriteria;
+  deadline: Date;
+  hashes: string[];
+  logs: RpcLogOutput[];
+  subscription: boolean;
+}
+
+export function bloomFilter(
+  bloom: Bloom,
+  addresses: Buffer[],
+  normalizedTopics: Array<Array<Buffer | null> | null>
+): boolean {
+  if (addresses.length > 0) {
+    let included = false;
+    for (const address of addresses) {
+      if (bloom.check(address)) {
+        included = true;
+        break;
+      }
+    }
+
+    if (!included) {
+      return false;
+    }
+  }
+
+  for (const sub of normalizedTopics) {
+    if (sub == null || sub.length === 0) {
+      continue;
+    }
+
+    let included = false;
+    for (const topic of sub) {
+      if (topic != null && bloom.check(topic)) {
+        included = true;
+        break;
+      }
+    }
+
+    if (!included) {
+      return false;
+    }
+  }
+  return true;
+}
+
+export function filterLogs(
+  logs: RpcLogOutput[],
+  criteria: FilterCriteria
+): RpcLogOutput[] {
+  const filteredLogs: RpcLogOutput[] = [];
+  for (const log of logs) {
+    const blockNumber = new BN(toBuffer(log.blockNumber!));
+    if (blockNumber.lt(criteria.fromBlock)) {
+      continue;
+    }
+
+    if (
+      !criteria.toBlock.eq(LATEST_BLOCK) &&
+      blockNumber.gt(criteria.toBlock)
+    ) {
+      continue;
+    }
+
+    if (
+      criteria.addresses.length !== 0 &&
+      !includes(criteria.addresses, toBuffer(log.address))
+    ) {
+      continue;
+    }
+
+    if (!topicMatched(criteria.normalizedTopics, log.topics)) {
+      continue;
+    }
+
+    filteredLogs.push(log);
+  }
+
+  return filteredLogs;
+}
+
+export function includes(addresses: Buffer[], a: Buffer): boolean {
+  for (const address of addresses) {
+    if (Buffer.compare(address, a) === 0) {
+      return true;
+    }
+  }
+
+  return false;
+}
+
+export function topicMatched(
+  normalizedTopics: Array<Array<Buffer | null> | null>,
+  logTopics: string[]
+): boolean {
+  for (let i = 0; i < normalizedTopics.length; i++) {
+    if (normalizedTopics.length > logTopics.length) {
+      return false;
+    }
+
+    const sub = normalizedTopics[i];
+    if (sub == null || sub.length === 0) {
+      continue;
+    }
+
+    let match: boolean = false;
+    for (const topic of sub) {
+      if (topic === null || logTopics[i] === bufferToHex(topic)) {
+        match = true;
+        break;
+      }
+    }
+    if (!match) {
+      return false;
+    }
+  }
+
+  return true;
+}

packages/buidler-core/src/internal/buidler-evm/provider/input.ts

@@ -60,6 +60,21 @@ export const rpcAddress = new t.Type<Buffer>(
   t.identity
 );
 
+export const logAddress = t.union([
+  rpcAddress,
+  t.array(rpcAddress),
+  t.undefined
+]);
+
+export type LogAddress = t.TypeOf<typeof logAddress>;
+
+export const logTopics = t.union([
+  t.array(t.union([t.null, rpcHash, t.array(t.union([t.null, rpcHash]))])),
+  t.undefined
+]);
+
+export type LogTopics = t.TypeOf<typeof logTopics>;
+
 export const optionalBlockTag = t.union([
   rpcQuantity,
   t.keyof({
@@ -120,6 +135,43 @@ export interface RpcCallRequestInput {
 
 export type RpcCallRequest = t.TypeOf<typeof rpcCallRequest>;
 
+export const rpcFilterRequest = t.type(
+  {
+    fromBlock: optionalBlockTag,
+    toBlock: optionalBlockTag,
+    address: logAddress,
+    topics: logTopics,
+    blockHash: optional(rpcHash)
+  },
+  "RpcFilterRequest"
+);
+
+export interface RpcSubscribe {
+  request: RpcFilterRequest;
+}
+
+export type OptionalRpcFilterRequest = t.TypeOf<
+  typeof optionalRpcFilterRequest
+>;
+
+export const optionalRpcFilterRequest = t.union([
+  rpcFilterRequest,
+  t.undefined
+]);
+
+export type RpcSubscribeRequest = t.TypeOf<typeof rpcSubscribeRequest>;
+
+export const rpcSubscribeRequest = t.keyof(
+  {
+    newHeads: null,
+    newPendingTransactions: null,
+    logs: null
+  },
+  "RpcSubscribe"
+);
+
+export type RpcFilterRequest = t.TypeOf<typeof rpcFilterRequest>;
+
 export function validateParams(params: any[]): [];
 
 export function validateParams(
@@ -200,6 +252,17 @@ export function validateParams(
   data: typeof rpcUnknown
 ): [Buffer, any];
 
+export function validateParams(
+  params: any[],
+  filterRequest: typeof rpcFilterRequest
+): [RpcFilterRequest];
+
+export function validateParams(
+  params: any[],
+  subscribeRequest: typeof rpcSubscribeRequest,
+  optionalFilterRequest: typeof optionalRpcFilterRequest
+): [RpcSubscribeRequest, OptionalRpcFilterRequest];
+
 export function validateParams(params: any[], number: typeof rpcQuantity): [BN];
 
 // tslint:disable only-buidler-error