Ethereum JSON RPC allows one to listen to contract events through eth_newFilter call. Based on non-authoritative sources, eth_newFilter takes topics parameter which is somehow a hash that is calculated from the event name and signature.
Where is explained what is topics parameter and how this should be calculated?
How to calculate topics parameter for a known Solidy event signature in Python?
Here is an example of a listener:
"""Poll Ethereum blockchain, install log hooks to call contracts.
Using geth JSON RPC API: https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newfilter
Copyright 2016 Mikko Ohtamaa - Licensed under MIT license.
"""
import logging
from typing import Callable, Iterable, List, Optional
import datetime
from .ethjsonrpc import EthJsonRpc
from .utils import sha3
#: Default logger
_logger = logging.getLogger(__name__)
#: Called when we spot a fired event. Callback is (contract_address, event_signature, api_data)
callback_type = Callable[[str, str, dict], None]
class ContractStatus:
"""Hold information about the processing status of a single contract."""
def __init__(self, filter_id, last_updated_at):
self.filter_id = filter_id
self.last_updated_at = last_updated_at
def now() -> datetime.datetime:
"""Get the current time as timezone-aware UTC timestamp."""
return datetime.datetime.now(datetime.timezone.utc)
def calculate_event_signature(decl: str) -> str:
"""Calculate bytecode signature of an event Solidy declaration.
Example:
.. code-block:
assert calculate_event_signature("VerifyTokenSet(address,uint256)") == "3D2E225F28C7AAA8014B84B0DD267E297CB25A0B24CB02AB9C9FCF76F660F05F"
To verify signature from the contract push opcodes:
.. code-block:: console
solc contract.sol --asm
To debug transactions on Morden testnet
* https://morden.ether.camp/transaction/9685191fece0dd0ef5a02210a305738be3fceb4089003924bc53de0cce0c0103
http://solidity.readthedocs.io/en/latest/contracts.html#events
https://github.com/ethereum/wiki/wiki/Solidity-Features#events
"""
return "0x" + sha3(decl.encode("utf-8")).hex().lower()
class ContractListener:
"""Fetch updates on events Solidy contract posts to Ethereum blockchain.
"""
def __init__(self, client: EthJsonRpc, events: Iterable[str], callback: callback_type, logger=_logger):
"""Create a contract listener.
Callbacks look like:
.. code-block:: python
def cb(address, event, api_data)
pass
:param client: EthJsonRpc instance we use to connect to geth node
:param events: List of Solidy event signatures we want to listne like like ``["Transfer(address,address,uint256)]``
:param callback: Callable that's going to get called for every new event detected.
:param logger: Optional
"""
self.logger = _logger
self.client = client
self.events = events
self.callback = callback
self.event_signatures = {calculate_event_signature(e): e for e in events}
#: Mapping contract address -> ContractStatus
self.currently_monitored_contracts = {}
def install_filter(self, contract_address: str):
"""Set up event filtering for a single contract using eth_newFilter.
:param contract_address: hex string
"""
installed_filter_id = self.client.eth_newFilter(from_block=0, address=contract_address)
status = ContractStatus(filter_id=installed_filter_id, last_updated_at=None)
self.currently_monitored_contracts[contract_address] = status
def process_events(self, status: ContractStatus, changes: Optional[List[dict]]) -> int:
updates = 0
# Nothing changed
if changes is None:
return 0
for change in changes:
contract_address = change["address"]
if contract_address not in self.currently_monitored_contracts:
self.logger.warn("Received a change for non-monitored contract %s", contract_address)
continue
topics = change["topics"]
if not topics:
self.logger.warn("Did not get topics with change data %s", change)
continue
event_type = topics[0]
if event_type not in self.event_signatures:
self.logger.warn("Unknown event signature %s", change)
continue
try:
self.callback(contract_address, self.event_signatures[event_type], change)
updates += 1
except Exception as e:
# IF we have bad code for processing one contract, don't stop at that but keep pushing for others
self.logger.error("Failed to update contract %s", contract_address)
self.logger.exception(e)
status.last_updated_at = now()
return updates
def fetch_all(self, contract: str) -> int:
status = self.currently_monitored_contracts[contract]
filter_id = status.filter_id
# Signature different as for newFilter :(
changes = self.client.eth_getLogs(dict(fromBlock=0, address=contract))
return self.process_events(status, changes)
def fetch_changes(self, contract) -> int:
"""Fetch latest events from geth.
.. note ::
The some transction might be posted twice due to ramp up and poll calls running differently.
Always make sure callbacks handle this.
:param contracts: List of contract addresses as hex string we are interested in
:return: Number of callbacks made
"""
status = self.currently_monitored_contracts[contract]
filter_id = status.filter_id
changes = self.client.eth_getFilterChanges(filter_id=filter_id)
return self.process_events(status, changes)
def monitor_contract(self, contract_address) -> int:
"""Start monitoring a contract and run callback for its all past events.
If contract is already added do nothing.
:param contract_address:
:return: Number of triggered callbacks
"""
assert type(contract_address) == str
assert contract_address.startswith("0x")
contract_address = contract_address.lower()
if contract_address in self.currently_monitored_contracts:
return 0
self.install_filter(contract_address)
return self.fetch_all(contract_address)
def remove_contract(self, contract_address):
del self.currently_monitored_contracts["contract_address"]
def poll(self):
"""Fetch changes to all monitored contracts.
Note that some events might be posted twice due to time elapse between ``monitor_contract`` and ``poll``.
:return: Number of triggered callbacks
"""
updates = 0
for c in self.currently_monitored_contracts.keys():
updates += self.fetch_changes(c)
return updates
Here is an example of usage:
import logging
import transaction
from sqlalchemy.orm import Session
from shareregistry.contractlistener import ContractListener
from shareregistry.models import Entity
from shareregistry.utils import bin_to_eth_address, eth_address_to_bin, txid_to_bin
from .ethjsonrpc import EthJsonRpc
logger = logging.getLogger(__name__)
class EntityUpdater:
def __init__(self, client: EthJsonRpc, dbsession: Session):
self.client = client
self.dbsession = dbsession
self.events = {
"created": "Created(address,uint256,string,string)",
"transfer": "Transfer(address,address,uint256)",
"verify": "VerifyTokenSet(address,uint256)",
}
self.contract_listener = ContractListener(self.client, self.events.values(), self.on_blockchain_event)
def on_blockchain_event(self, address: str, event: str, api_data: dict):
"""Called by ContractLister to tell us about the contract events.
:param address: Contract address as a hex string
:param event: One of event signatures
:param api_data: eth_getFilterChanges() result https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterchanges
:return:
"""
logger.info("Received contract event %s for %s", event, address)
bin_address = eth_address_to_bin(address)
with transaction.manager:
entity = self.dbsession.query(Entity).filter_by(contract_address=bin_address).one()
if event == self.events["created"]:
txid = txid_to_bin(api_data["transactionHash"])
entity.initial_supply = 0
entity.txid = txid
elif event == self.events["transfer"]:
from_address = api_data["topics"][1]
to_address = api_data["topics"][2]
value = int(api_data["data"][0])
txid = api_data["transactionHash"]
else:
raise RuntimeError("Unknown event: {}".format(event))
def update_all(self) -> int:
"""Poll geth and get updates for the contracts in our database."""
updates = 0
contracts = []
# Get list of all known contracts in the database
with transaction.manager:
for e in self.dbsession.query(Entity).all():
if not e.contract_address:
continue
contracts.append(bin_to_eth_address(e.contract_address))
# Each callback will run in its own db transaction context
for c in contracts:
updates += self.contract_listener.monitor_contract(c)
updates += self.contract_listener.poll()
return updates
solc contract.sol --asmand check values of event push opcodes