4

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?

3

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
  • What if eventually "Failed to update contract" exception raises and some changes wouldn't be written to you local DB? Wouldn't that mean that you cannot use eth_getFilterChanges again as the blockchain would think you've already got those changes? How do you handle this situation? Or is there something I've missed (in code or in how filters work)? – takeshi Aug 16 '16 at 12:17
  • I did not understand your question. Please refer to full source code github.com/websauna/websauna.wallet/blob/feat/ethereum-3/… – Mikko Ohtamaa Aug 16 '16 at 23:58
  • I see you made it in a different way now. I just wanted to understand how eth_getFilterChanges works. AFAIU it returns an event only once. So one needs to make sure his application will not "loose" that event. And if it will, the application will have to get that event with another method (like web3.get_logs for example). – takeshi Aug 24 '16 at 13:13
  • 1
    @takeshi: I recommend not doing stateful (new filter, get filter changes), but manage all the filter state in your application code and only use getLogs. – Mikko Ohtamaa Aug 25 '16 at 20:16
  • 1
    @takeshi: You have more control and in doubt you can just reread the events and update your application database. – Mikko Ohtamaa Aug 25 '16 at 20:16

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.