Refactors Subscriptions to be Constant (#140)

Constant loading of Subscriptions

chains/ethereum/config.go

@@ -15,32 +15,30 @@ import (
 
 // Config encapsulates all necessary parameters in ethereum compatible forms
 type Config struct {
-	id            msg.ChainId        // ChainID
-	chainID       *big.Int           // Ethereum chain ID
-	endpoint      string             // url for rpc endpoint
-	from          string             // address of key to use
-	subscriptions []string           // list of events to subscribe to
-	keystore      *keystore.Keystore // Location of keyfiles
-	receiver      common.Address
-	emitter       common.Address
-	gasLimit      *big.Int
-	gasPrice      *big.Int
+	id       msg.ChainId        // ChainID
+	chainID  *big.Int           // Ethereum chain ID
+	endpoint string             // url for rpc endpoint
+	from     string             // address of key to use
+	keystore *keystore.Keystore // Location of keyfiles
+	receiver common.Address
+	emitter  common.Address
+	gasLimit *big.Int
+	gasPrice *big.Int
 }
 
 // ParseChainConfig uses a core.ChainConfig to construct a corresponding Config
 func ParseChainConfig(chainCfg *core.ChainConfig) (*Config, error) {
 
 	config := &Config{
-		id:            chainCfg.Id,
-		endpoint:      chainCfg.Endpoint,
-		from:          chainCfg.From,
-		subscriptions: chainCfg.Subscriptions,
-		keystore:      chainCfg.Keystore,
-		chainID:       big.NewInt(1),
-		receiver:      common.HexToAddress("0x0"),
-		emitter:       common.HexToAddress("0x0"),
-		gasLimit:      big.NewInt(6721975),
-		gasPrice:      big.NewInt(20000000000),
+		id:       chainCfg.Id,
+		endpoint: chainCfg.Endpoint,
+		from:     chainCfg.From,
+		keystore: chainCfg.Keystore,
+		chainID:  big.NewInt(1),
+		receiver: common.HexToAddress("0x0"),
+		emitter:  common.HexToAddress("0x0"),
+		gasLimit: big.NewInt(6721975),
+		gasPrice: big.NewInt(20000000000),
 	}
 
 	if chainID, ok := chainCfg.Opts["chainID"]; ok {

chains/ethereum/connection.go

@@ -69,14 +69,14 @@ func (c *Connection) NetworkId() (*big.Int, error) {
 }
 
 // subscribeToEvent registers an rpc subscription for the event with the signature sig for contract at address
-func (c *Connection) subscribeToEvent(query eth.FilterQuery) (*Subscription, error) {
+func (c *Connection) subscribeToEvent(query eth.FilterQuery) (*ActiveSubscription, error) {
 	ch := make(chan ethtypes.Log)
 	sub, err := c.conn.SubscribeFilterLogs(c.ctx, query, ch)
 	if err != nil {
 		close(ch)
 		return nil, err
 	}
-	return &Subscription{
+	return &ActiveSubscription{
 		ch:  ch,
 		sub: sub,
 	}, nil

chains/ethereum/events.go

@@ -14,6 +14,17 @@ import (
 	ethcrypto "github.com/ethereum/go-ethereum/crypto"
 )
 
+const (
+	DepositAsset = "DepositAsset"
+	NftTransfer  = "NftTransfer"
+	ErcTransfer  = "ErcTransfer"
+
+	// For internal usage only
+	DepositAssetSignature = "DepositAsset(address,bytes32)"
+	NftTransferSignature  = "NFTTransfer(uint256,uint256,address,address,uint256,bytes)"
+	ErcTransferSignature  = "ERCTransfer(uint256,uint256,address,uint256,address)"
+)
+
 func (l *Listener) handleTransferEvent(eventI interface{}) msg.Message {
 	log15.Debug("Handling deposit proposal event")
 	event := eventI.(ethtypes.Log)

chains/ethereum/listener.go

@@ -4,7 +4,6 @@
 package ethereum
 
 import (
-	"fmt"
 	"math/big"
 
 	"github.com/ChainSafe/ChainBridgeV2/chains"
@@ -18,14 +17,19 @@ import (
 var _ chains.Listener = &Listener{}
 
 type Subscription struct {
+	signature string
+	handler   chains.EvtHandlerFn
+}
+
+type ActiveSubscription struct {
 	ch  <-chan ethtypes.Log
 	sub eth.Subscription
 }
 
 type Listener struct {
 	cfg             Config
 	conn            *Connection
-	subscriptions   map[EventSig]*Subscription
+	subscriptions   map[EventSig]*ActiveSubscription
 	router          chains.Router
 	emitterContract EmitterContract // instance of bound emitter contract
 }
@@ -34,7 +38,7 @@ func NewListener(conn *Connection, cfg *Config) *Listener {
 	return &Listener{
 		cfg:           *cfg,
 		conn:          conn,
-		subscriptions: make(map[EventSig]*Subscription),
+		subscriptions: make(map[EventSig]*ActiveSubscription),
 	}
 }
 
@@ -46,24 +50,32 @@ func (l *Listener) SetRouter(r chains.Router) {
 	l.router = r
 }
 
+func (l *Listener) GetSubscriptions() []*Subscription {
+	return []*Subscription{
+		{
+			signature: ErcTransfer,
+			handler:   l.handleTransferEvent,
+		},
+		{
+			signature: NftTransfer,
+			handler:   l.handleTransferEvent,
+		},
+		{
+			signature: DepositAsset,
+			handler:   l.handleTestDeposit,
+		},
+	}
+
+}
+
 // Start registers all subscriptions provided by the config
 func (l *Listener) Start() error {
-	log15.Debug("Starting listener...", "chainID", l.cfg.id, "subs", l.cfg.subscriptions)
-	for _, subscription := range l.cfg.subscriptions {
-		sub := subscription
-		switch sub {
-		case ErcTransferSignature, NftTransferSignature:
-			err := l.RegisterEventHandler(sub, l.handleTransferEvent)
-			if err != nil {
-				log15.Error("failed to register event handler", "err", err)
-			}
-		case DepositAssetSignature:
-			err := l.RegisterEventHandler(sub, l.handleTestDeposit)
-			if err != nil {
-				log15.Error("failed to register event handler", "err", err)
-			}
-		default:
-			return fmt.Errorf("Unrecognized event: %s", sub)
+	log15.Debug("Starting listener...", "chainID", l.cfg.id)
+	subscriptions := l.GetSubscriptions()
+	for _, sub := range subscriptions {
+		err := l.RegisterEventHandler(sub.signature, sub.handler)
+		if err != nil {
+			log15.Error("failed to register event handler", "err", err)
 		}
 	}
 	return nil
@@ -100,7 +112,7 @@ func (l *Listener) RegisterEventHandler(subscription string, handler chains.EvtH
 
 // watchEvent will call the handler for every occurrence of the corresponding event. It should be run in a separate
 // goroutine to monitor the subscription channel.
-func (l *Listener) watchEvent(eventSubscription *Subscription, handler func(interface{}) msg.Message) {
+func (l *Listener) watchEvent(eventSubscription *ActiveSubscription, handler func(interface{}) msg.Message) {
 	for {
 		select {
 		case evt := <-eventSubscription.ch:

cmd/chainbridge/main.go

@@ -137,20 +137,16 @@ func run(ctx *cli.Context) error {
 				Id:       chain.Id,
 				Endpoint: chain.Endpoint,
 				From:     chain.From,
-				// TODO remove this in favour of OPTS when config PR lands
-				Subscriptions: ethereum.BuildEventSubscriptions([]string{"DepositAsset", "NftTransfer", "ErcTransfer"}),
-				Keystore:      ks,
-				Opts:          chain.Opts,
+				Keystore: ks,
+				Opts:     chain.Opts,
 			})
 		} else if chain.Type == "substrate" {
 			chainconfig, err = ethereum.InitializeChain(&core.ChainConfig{
 				Id:       chain.Id,
 				Endpoint: chain.Endpoint,
 				From:     chain.From,
-				// TODO remove this in favour of OPTS when config PR lands
-				Subscriptions: ethereum.BuildEventSubscriptions([]string{"DepositAsset", "NftTransfer", "ErcTransfer"}),
-				Keystore:      ks,
-				Opts:          chain.Opts,
+				Keystore: ks,
+				Opts:     chain.Opts,
 			})
 		} else {
 			return errors.New("Unrecognized Chain Type")