feat: tx-filtering

README.md

@@ -32,6 +32,39 @@ except Exception as e:
   print("error subscribing", e)
 ```
 
+### Filtering
+The first argument to `subscribe_new_txs` is a filter, which can be empty if you want to get all transactions. A filter can be built with the filter package
+```python
+try:
+  # Construct filter
+  # example 1: all transactions with either of these addresses as the receiver
+  f = filter.Filter(
+          filter.filter_or([filter.filter_to("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"), filter.filter_to("0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D")])
+  )
+
+  # example 2: all ERC20 transfers on the 2 tokens below
+    f = filter.Filter(filter.filter_and(
+        filter.method_id("0xa9059cbb"),
+        filter.filter_or(
+            filter.filter_to("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"),
+            filter.filter_to("0xdAC17F958D2ee523a2206206994597C13D831ec7"),
+        ),
+    ))
+
+  sub = client.subscribe_new_txs(f)
+
+  # Iterate over transaction stream
+  for tx in sub:
+    do_something(tx)
+except Exception as e:
+  print("error subscribing", e)
+```
+You can currently filter the following properties
+* To
+* From
+* MethodID
+* Value (greater than)
+
 **Transaction type**:
 We export our own transaction type. All the bytes fields are encoded as hexadecimal strings.
 ```python

fiber/client.py

@@ -1,6 +1,7 @@
 from fiber import eth_pb2
 from fiber import api_pb2_grpc
 from fiber import api_pb2
+from fiber.filter import filter
 import grpc
 
 from eth_typing import Address
@@ -138,8 +139,8 @@ def connect(self):
         channel = grpc.insecure_channel(self.api)
         self.stub = api_pb2_grpc.APIStub(channel)
 
-    def subscribe_new_txs(self):
-        return map(lambda proto: proto_to_tx(proto), self.stub.SubscribeNewTxs(api_pb2.TxFilter(), metadata=self.metadata))
+    def subscribe_new_txs(self, fil: filter.Filter = None):
+        return map(lambda proto: proto_to_tx(proto), self.stub.SubscribeNewTxs(api_pb2.TxFilter(bytes(fil.encode())), metadata=self.metadata))
 
     def subscribe_new_blocks(self):
         return map(lambda proto: proto_to_block(proto), self.stub.SubscribeNewBlocks(api_pb2.google_dot_protobuf_dot_empty__pb2.Empty(), metadata=self.metadata))

fiber/filter/filter.py

@@ -0,0 +1,112 @@
+from fiber import eth_pb2
+from fiber import api_pb2_grpc
+from fiber import api_pb2
+import grpc
+import json
+
+from eth_utils import decode_hex
+from typing import Any
+
+FILTER_AND = 0
+FILTER_OR = 1
+
+class FilterKv:
+    key: str
+    value: str
+
+    def __init__(self, key: str, value: str):
+        self.key = key
+        self.value = value
+
+
+class Node:
+    operand: FilterKv
+    operator: int
+    children: list
+
+    def __init__(self, operand: int, operator: int, children: list):
+        self.operand = operand
+        self.operator = operator
+        self.children = children
+
+class Filter:
+    root: Node
+
+    def __init__(self, root: Node):
+        self.root = root
+
+    def encode(self) -> str:
+        return json.JSONEncoder.encode(self)
+
+
+def filter_and(ops: list) -> callable[[Filter, Node], None]:
+    def f(f, n):
+        new = {}
+
+        if n == None:
+            new = Node(None, FILTER_OR, None)
+
+            f.root = new
+        else:
+            new = Node(None, FILTER_OR, None)
+
+            n.children.append(new)
+
+        for op in ops:
+            op(f, n)
+
+    return f
+
+def filter_or(ops: list) -> callable[[Filter, Node], None]:
+    def f(f, n):
+        new = {}
+
+        if n == None:
+            new = Node(None, FILTER_AND, None)
+
+            f.root = new
+        else:
+            new = Node(None, FILTER_AND, None)
+
+            n.children.append(new)
+
+        for op in ops:
+            op(f, n)
+
+    return f
+
+def filter_to(to: str) -> callable[[Filter, Node], None]:
+    def f (f, n):
+        if n == None:
+            f.root = Node(FilterKv("to", decode_hex(to)), None, None)
+        else:
+            n.children.append(Node(FilterKv("to", decode_hex(to)), None, None))
+
+    return f
+
+def filter_from(from_: str) -> callable[[Filter, Node], None]:
+    def f (f, n):
+        if n == None:
+            f.root = Node(FilterKv("from", decode_hex(from_)), None, None)
+        else:
+            n.children.append(Node(FilterKv("from", decode_hex(from_)), None, None))
+
+    return f
+
+def method_id(id: str) -> callable[[Filter, Node], None]:
+    def f (f, n):
+        if n == None:
+            f.root = Node(FilterKv("method_id", id), None, None)
+        else:
+            n.children.append(Node(FilterKv("method_id", id), None, None))
+
+    return f
+
+def filter_value(value: int) -> callable[[Filter, Node], None]:
+    def f (f, n):
+        if n == None:
+            f.root = Node(FilterKv("value", str(value)), None, None)
+        else:
+            n.children.append(Node(FilterKv("value", str(value)), None, None))
+
+    return f

fiber/filter/test_filter.py

@@ -0,0 +1,18 @@
+import unittest
+import json
+from fiber.filter import filter
+
+class TestFilter(unittest.TestCase):
+    def test_or_filter(self):
+        f = filter.Filter(filter.filter_or([filter.filter_to("0x1"), filter.filter_to("0x2")]))
+
+        print(f.encode())
+
+    def test_and_filter(self):
+        f = filter.Filter(filter.filter_and([filter.filter_to("0x1"), filter.filter_to("0x2")]))
+
+        print(f.encode())
+
+
+if __name__ == '__main__':
+    unittest.main()