Implement a Shard Transaction Generator

[rauljordan]

Now that #149 is merged, we can create a simple shard tx generator as a service attached to ShardEthereum from sharding/node/backend.go that publishes []*types.Transaction data to a go channel.

The proposer service will need this in order to process these incoming transactions into serialized blobs that will then be packaged into collations.

[rauljordan]

Hey @rawfalafel, as I mentioned, I started on this issue but happy for you to take over if you want.

The idea was that I added an event feed to the shard txpool as follows:

in sharding/txpool/service.go

// ShardTXPool handles a transaction pool for a sharded system.
type ShardTXPool struct {
	p2p              sharding.ShardP2P
	transactionsFeed *event.Feed
}

// NewShardTXPool creates a new observer instance.
func NewShardTXPool(p2p sharding.ShardP2P) (*ShardTXPool, error) {
	return &ShardTXPool{p2p: p2p, transactionsFeed: new(event.Feed)}, nil
}

// Start the main routine for a shard transaction pool.
func (pool *ShardTXPool) Start() error {
	log.Info("Starting shard txpool service")
	go pool.generateTestTransactions()
	return nil
}

func (pool *ShardTXPool) TransactionsFeed() *event.Feed {
	return pool.transactionsFeed
}

func (pool *ShardTXPool) generateTestTransactions() {
	for {
		nsent := pool.transactionsFeed.Send(1)
		log.Info(fmt.Sprintf("Sent transaction to %d subscribers", nsent))
		time.Sleep(time.Second)
	}
}

and then in the proposer subpackage, I subscribed to this feed via a for, select block:

in sharding/proposer/service.go

// Start the main loop for proposing collations.
func (p *Proposer) Start() error {
	log.Info("Starting proposer service")
	go p.subscribeTransactions()
	return nil
}

func (p *Proposer) subscribeTransactions() {
	// Subscribes to incoming transactions from the txpool via the shardp2p network.
	for {
		subchan := make(chan int)
		sub := p.txpool.TransactionsFeed().Subscribe(subchan)
		// 10 second time out for the subscription.
		timeout := time.NewTimer(10 * time.Second)
		select {
		case v := <-subchan:
			log.Info(fmt.Sprintf("Received transaction with id: %d", v))
		case <-timeout.C:
			log.Error("Receive timeout")
		}

		sub.Unsubscribe()
		select {
		case _, ok := <-sub.Err():
			if ok {
				log.Error("Channel not closed after unsubscribe")
			}
		case <-timeout.C:
			log.Error("Unsubscribe timeout")
		}
	}
}

Right now I just used an int for the channel but we can make this system more elaborate by generating a variety of different transactions. The PR should also include tests and a nicer way to generate them at random intervals + handling timeouts. Let me know if you are interested. I started this on my personal branch but happy for you to take the initial code from there and take over.

[rauljordan]

Assigning to @rawfalafel (Yutaro)

[rawfalafel]

Pulled in your branch and taking a look now. Thanks @rauljordan !