package module
Version: v0.0.0-...-98400ad Latest Latest

This package is not in the latest version of its module.

Go to latest
Published: Sep 13, 2020 License: Apache-2.0 Imports: 11 Imported by: 2



ethereum-watcher is an event listener for the Ethereum Blockchain written in Golang. With ethereum-watcher you can monitor and track current or historic events that occur on the Ethereum Blockchain.


Many applications that interact with the Ethereum Blockchain need to know when specific actions occur on the chain, but cannot directly access the on-chain data. ethereum-watcher acts as an interface between application and chain: gathering specified data from the blockchain so that applications can more seamlessly interact with on-chain events.


  1. Plug-in friendly. You can easily add a plugin to ethereum-watcher to listen to any type of on-chain event.
  2. Fork Tolerance. If a fork occurs, a revert message is sent to the subscriber.

Example Use Cases

  • DefiWatch monitors the state of Ethereum addresses on several DeFi platforms including DDEX, Compound, DyDx, and Maker. It tracks things like loan ROI, current borrows, liquidations, etc. To track all of this on multiple platforms, DefiWatch has to continuously receive updates from their associated smart contracts. This is done using ethereum-watcher instead of spending time dealing with serialization/deserialization messages from the Ethereum Node, so they can focus on their core logic.
  • Profit & Loss calculations on DDEX. DDEX provides their margin trading users with estimated Profit and Loss (P&L) calculations for their margin positions. To update the P&L as timely and accurately as possible, DDEX uses ethereum-watcher to listen to updates from the Ethereum Blockchain. These updates include: onchain price updates, trading actions from users, and more.
  • DDEX also uses an "Eth-Transaction-Watcher" to monitor the on-chain status of trading transactions. DDEX needs to know the latest states of these transactions once they are included in newly mined blocks, so that the platform properly updates trading balances and histories. This is done using the TxReceiptPlugin of ethereum-watcher.


Run go get

Sample Commands

This project is primarily designed as a library to build upon. However, to help others easily understand how ethereum-watcher works and what it is capable of, we prepared some sample commands for you to try out.

display basic help info

docker run hydroprotocolio/ethereum-watcher:master /bin/ethereum-watcher help

ethereum-watcher makes getting updates from Ethereum easier

  ethereum-watcher [command]

Available Commands:
  contract-event-listener listen and print events from contract
  help                    Help about any command
  new-block-number        Print number of new block
  usdt-transfer           Show Transfer Event of USDT

  -h, --help   help for ethereum-watcher

Use "ethereum-watcher [command] --help" for more information about a command.

print new block numbers

docker run hydroprotocolio/ethereum-watcher:master /bin/ethereum-watcher new-block-number

time="2020-01-07T07:33:17Z" level=info msg="waiting for new block..."
time="2020-01-07T07:33:19Z" level=info msg=">> found new block: 9232152, is removed: false"
time="2020-01-07T07:33:44Z" level=info msg=">> found new block: 9232153, is removed: false"
time="2020-01-07T07:34:03Z" level=info msg=">> found new block: 9232154, is removed: false"
time="2020-01-07T07:34:04Z" level=info msg=">> found new block: 9232155, is removed: false"
time="2020-01-07T07:34:05Z" level=info msg=">> found new block: 9232156, is removed: false"

see USDT transfer events

docker run hydroprotocolio/ethereum-watcher:master /bin/ethereum-watcher usdt-transfer

time="2020-01-07T07:34:32Z" level=info msg="See new USDT Transfer at block: 9232158, count:  9"
time="2020-01-07T07:34:32Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:32Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:32Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:32Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:32Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:32Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:32Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:32Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:32Z" level=info msg="  >> tx:"

time="2020-01-07T07:34:50Z" level=info msg="See new USDT Transfer at block: 9232159, count: 18"
time="2020-01-07T07:34:50Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:50Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:50Z" level=info msg="  >> tx:"
time="2020-01-07T07:34:50Z" level=info msg="  >> tx:"

see specific events that occur within a smart contract. The example shows Transfer & Approve events from Multi-Collateral-DAI

docker run hydroprotocolio/ethereum-watcher:master /bin/ethereum-watcher contract-event-listener \
    --block-backoff 100 \
    --contract 0x6b175474e89094c44da98b954eedeac495271d0f \
    --events 0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925 0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef
INFO[2020-01-07T18:05:26+08:00] --block-backoff activated, we start from block: 9232741 (= 9232841 - 100)

INFO[2020-01-07T18:05:27+08:00] # of interested events at block(9232741->9232745): 1
INFO[2020-01-07T18:05:27+08:00]   >> tx:

INFO[2020-01-07T18:05:28+08:00] # of interested events at block(9232746->9232750): 3
INFO[2020-01-07T18:05:28+08:00]   >> tx:
INFO[2020-01-07T18:05:28+08:00]   >> tx:
INFO[2020-01-07T18:05:28+08:00]   >> tx:

INFO[2020-01-07T18:05:29+08:00] # of interested events at block(9232751->9232755): 2
INFO[2020-01-07T18:05:29+08:00]   >> tx:
INFO[2020-01-07T18:05:29+08:00]   >> tx:

Here the flag --block-backoff signals for ethereum-watcher to use historic tracking from 100 blocks ago.


To effectively use ethereum-watcher, you will be interacting with two primary structs:

  • Watcher
  • ReceiptLogWatcher


Watcher is an HTTP client which continuously polls newly mined blocks on the Ethereum Blockchain. We can incorporate various kinds of "plugins" into Watcher, which will poll for specific types of events and data, such as:

  • BlockPlugin
  • TransactionPlugin
  • TransactionReceiptPlugin
  • ReceiptLogPlugin

Once the Watcher sees a new block, it will parse the info and feed the data into the registered plugins. You can see some of the code examples below.

The plugins are designed to be easily modifiable so that you can create your own plugin based on the provided ones. For example, the code shown below registers an ERC20TransferPlugin to show new ERC20 Transfer Events. This plugin simply parses some receipt info from a different TransactionReceiptPlugin. So if you want to show more info than what the ERC20TransferPlugin shows, like the gas used in the transaction, you can easily create a BetterERC20TransferPlugin showing that.

Watcher Examples
Print number of newly mined blocks
package main

import (

func main() {
	api := ""
	w := NewHttpBasedEthWatcher(context.Background(), api)

	// we use BlockPlugin here
	w.RegisterBlockPlugin(plugin.NewBlockNumPlugin(func(i uint64, b bool) {
		fmt.Println(">>", i, b)

Listen for new ERC20 Transfer Events
package main

import (

func main() {
	api := ""
	w := NewHttpBasedEthWatcher(context.Background(), api)

	// we use TxReceiptPlugin here
		func(token, from, to string, amount decimal.Decimal, isRemove bool) {

			logrus.Infof("New ERC20 Transfer >> token(%s), %s -> %s, amount: %s, isRemoved: %t",
				token, from, to, amount, isRemove)




Watcher is polling for blocks one by one, so what if we want to query certain events from the latest 10000 blocks? Watcher can do that but fetching blocks one by one can be slow. ReceiptLogWatcher to the rescue!

ReceiptLogWatcher makes use of the eth_getLogs to query for logs in a batch. Check out the code below to see how to use it.

Example of ReceiptLogWatcher
package main

import (

func main() {
	api := ""
	usdtContractAdx := "0xdac17f958d2ee523a2206206994597c13d831ec7"
	// ERC20 Transfer Event
	topicsInterestedIn := []string{"0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"}

	handler := func(from, to int, receiptLogs []blockchain.IReceiptLog, isUpToHighestBlock bool) error {
		logrus.Infof("USDT Transfer count: %d, %d -> %d", len(receiptLogs), from, to)
		return nil

	// query for USDT Transfer Events
	receiptLogWatcher := NewReceiptLogWatcher(
			StepSizeForBigLag:               5,
			IntervalForPollingNewBlockInSec: 5,
			RPCMaxRetry:                     3,
			ReturnForBlockWithNoReceiptLog:  true,



Apache 2.0 License




View Source
const DefaultStepSizeForBigLag = 10


This section is empty.


func ListenForReceiptLogTillExit

func ListenForReceiptLogTillExit(
	ctx context.Context,
	api string,
	startBlock int,
	contract string,
	interestedTopics []string,
	handler func(receiptLog structs.RemovableReceiptLog),
) int

deprecated, please use receipt_log_watcher instead.


type AbstractWatcher

type AbstractWatcher struct {
	Ctx context.Context

	NewBlockChan        chan *structs.RemovableBlock
	NewTxAndReceiptChan chan *structs.RemovableTxAndReceipt
	NewReceiptLogChan   chan *structs.RemovableReceiptLog

	SyncedBlocks         *list.List
	SyncedTxAndReceipts  *list.List
	MaxSyncedBlockToKeep int

	BlockPlugins      []plugin.IBlockPlugin
	TxPlugins         []plugin.ITxPlugin
	TxReceiptPlugins  []plugin.ITxReceiptPlugin
	ReceiptLogPlugins []plugin.IReceiptLogPlugin

	ReceiptCatchUpFromBlock uint64
	// contains filtered or unexported fields

func NewHttpBasedEthWatcher

func NewHttpBasedEthWatcher(ctx context.Context, api string) *AbstractWatcher

func (*AbstractWatcher) FoundFork

func (watcher *AbstractWatcher) FoundFork(newBlock blockchain.Block) bool

func (*AbstractWatcher) LatestSyncedBlockNum

func (watcher *AbstractWatcher) LatestSyncedBlockNum() uint64

func (*AbstractWatcher) RegisterBlockPlugin

func (watcher *AbstractWatcher) RegisterBlockPlugin(plugin plugin.IBlockPlugin)

func (*AbstractWatcher) RegisterReceiptLogPlugin

func (watcher *AbstractWatcher) RegisterReceiptLogPlugin(plugin plugin.IReceiptLogPlugin)

func (*AbstractWatcher) RegisterTxPlugin

func (watcher *AbstractWatcher) RegisterTxPlugin(plugin plugin.ITxPlugin)

func (*AbstractWatcher) RegisterTxReceiptPlugin

func (watcher *AbstractWatcher) RegisterTxReceiptPlugin(plugin plugin.ITxReceiptPlugin)

func (*AbstractWatcher) RunTillExit

func (watcher *AbstractWatcher) RunTillExit() error

start sync from latest block

func (*AbstractWatcher) RunTillExitFromBlock

func (watcher *AbstractWatcher) RunTillExitFromBlock(startBlockNum uint64) error

start sync from given block 0 means start from latest block

func (*AbstractWatcher) SetSleepSecondsForNewBlock

func (watcher *AbstractWatcher) SetSleepSecondsForNewBlock(sec int)

type ReceiptLogWatcher

type ReceiptLogWatcher struct {
	// contains filtered or unexported fields

func NewReceiptLogWatcher

func NewReceiptLogWatcher(
	ctx context.Context,
	api string,
	startBlockNum int,
	contract string,
	interestedTopics []string,
	handler func(from, to int, receiptLogs []blockchain.IReceiptLog, isUpToHighestBlock bool) error,
	configs ...ReceiptLogWatcherConfig,
) *ReceiptLogWatcher

func (*ReceiptLogWatcher) GetHighestSyncedBlockNum

func (w *ReceiptLogWatcher) GetHighestSyncedBlockNum() int

func (*ReceiptLogWatcher) GetHighestSyncedBlockNumAndLogIndex

func (w *ReceiptLogWatcher) GetHighestSyncedBlockNumAndLogIndex() (int, int)

func (*ReceiptLogWatcher) Run

func (w *ReceiptLogWatcher) Run() error

type ReceiptLogWatcherConfig

type ReceiptLogWatcherConfig struct {
	StepSizeForBigLag               int
	ReturnForBlockWithNoReceiptLog  bool
	IntervalForPollingNewBlockInSec int
	RPCMaxRetry                     int
	LagToHighestBlock               int
	StartSyncAfterLogIndex          int

type SyncSignal

type SyncSignal struct {
	// contains filtered or unexported fields

func (*SyncSignal) Done

func (s *SyncSignal) Done()

func (*SyncSignal) Permit

func (s *SyncSignal) Permit()

func (*SyncSignal) WaitDone

func (s *SyncSignal) WaitDone()

func (*SyncSignal) WaitPermission

func (s *SyncSignal) WaitPermission()


Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL