use ethreact.Event and ethreact.ReactorEngine

ethchain/dagger.go

@@ -3,6 +3,7 @@ package ethchain
 import (
 	"github.com/ethereum/eth-go/ethcrypto"
 	"github.com/ethereum/eth-go/ethlog"
+	"github.com/ethereum/eth-go/ethreact"
 	"github.com/ethereum/eth-go/ethutil"
 	"github.com/obscuren/sha3"
 	"hash"
@@ -14,7 +15,7 @@ import (
 var powlogger = ethlog.NewLogger("POW")
 
 type PoW interface {
-	Search(block *Block, reactChan chan ethutil.React) []byte
+	Search(block *Block, reactChan chan ethreact.Event) []byte
 	Verify(hash []byte, diff *big.Int, nonce []byte) bool
 }
 
@@ -22,7 +23,7 @@ type EasyPow struct {
 	hash *big.Int
 }
 
-func (pow *EasyPow) Search(block *Block, reactChan chan ethutil.React) []byte {
+func (pow *EasyPow) Search(block *Block, reactChan chan ethreact.Event) []byte {
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 	hash := block.HashNoNonce()
 	diff := block.Difficulty

ethchain/state_manager.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/ethereum/eth-go/ethcrypto"
 	"github.com/ethereum/eth-go/ethlog"
+	"github.com/ethereum/eth-go/ethreact"
 	"github.com/ethereum/eth-go/ethtrie"
 	"github.com/ethereum/eth-go/ethutil"
 	"github.com/ethereum/eth-go/ethwire"
@@ -36,7 +37,7 @@ type EthManager interface {
 	BlockChain() *BlockChain
 	TxPool() *TxPool
 	Broadcast(msgType ethwire.MsgType, data []interface{})
-	Reactor() *ethutil.ReactorEngine
+	Reactor() *ethreact.ReactorEngine
 	PeerCount() int
 	IsMining() bool
 	IsListening() bool

ethereum.go

@@ -6,6 +6,7 @@ import (
 	"github.com/ethereum/eth-go/ethchain"
 	"github.com/ethereum/eth-go/ethcrypto"
 	"github.com/ethereum/eth-go/ethlog"
+	"github.com/ethereum/eth-go/ethreact"
 	"github.com/ethereum/eth-go/ethrpc"
 	"github.com/ethereum/eth-go/ethutil"
 	"github.com/ethereum/eth-go/ethwire"
@@ -71,7 +72,7 @@ type Ethereum struct {
 
 	listening bool
 
-	reactor *ethutil.ReactorEngine
+	reactor *ethreact.ReactorEngine
 
 	RpcServer *ethrpc.JsonRpcServer
 
@@ -106,7 +107,7 @@ func New(db ethutil.Database, clientIdentity ethwire.ClientIdentity, keyManager
 		keyManager:     keyManager,
 		clientIdentity: clientIdentity,
 	}
-	ethereum.reactor = ethutil.NewReactorEngine()
+	ethereum.reactor = ethreact.New()
 
 	ethereum.txPool = ethchain.NewTxPool(ethereum)
 	ethereum.blockChain = ethchain.NewBlockChain(ethereum)
@@ -118,7 +119,7 @@ func New(db ethutil.Database, clientIdentity ethwire.ClientIdentity, keyManager
 	return ethereum, nil
 }
 
-func (s *Ethereum) Reactor() *ethutil.ReactorEngine {
+func (s *Ethereum) Reactor() *ethreact.ReactorEngine {
 	return s.reactor
 }
 
@@ -350,6 +351,7 @@ func (s *Ethereum) ReapDeadPeerHandler() {
 
 // Start the ethereum
 func (s *Ethereum) Start(seed bool) {
+	s.reactor.Start()
 	// Bind to addr and port
 	ln, err := net.Listen("tcp", ":"+s.Port)
 	if err != nil {
@@ -461,6 +463,9 @@ func (s *Ethereum) Stop() {
 	s.txPool.Stop()
 	s.stateManager.Stop()
 
+	s.reactor.Flush()
+	s.reactor.Stop()
+
 	ethlogger.Infoln("Server stopped")
 	close(s.shutdownChan)
 }

ethminer/miner.go

@@ -4,7 +4,7 @@ import (
 	"bytes"
 	"github.com/ethereum/eth-go/ethchain"
 	"github.com/ethereum/eth-go/ethlog"
-	"github.com/ethereum/eth-go/ethutil"
+	"github.com/ethereum/eth-go/ethreact"
 	"github.com/ethereum/eth-go/ethwire"
 	"sort"
 )
@@ -15,19 +15,19 @@ type Miner struct {
 	pow         ethchain.PoW
 	ethereum    ethchain.EthManager
 	coinbase    []byte
-	reactChan   chan ethutil.React
+	reactChan   chan ethreact.Event
 	txs         ethchain.Transactions
 	uncles      []*ethchain.Block
 	block       *ethchain.Block
 	powChan     chan []byte
-	powQuitChan chan ethutil.React
+	powQuitChan chan ethreact.Event
 	quitChan    chan bool
 }
 
 func NewDefaultMiner(coinbase []byte, ethereum ethchain.EthManager) Miner {
-	reactChan := make(chan ethutil.React, 1)   // This is the channel that receives 'updates' when ever a new transaction or block comes in
-	powChan := make(chan []byte, 1)            // This is the channel that receives valid sha hases for a given block
-	powQuitChan := make(chan ethutil.React, 1) // This is the channel that can exit the miner thread
+	reactChan := make(chan ethreact.Event, 1)   // This is the channel that receives 'updates' when ever a new transaction or block comes in
+	powChan := make(chan []byte, 1)             // This is the channel that receives valid sha hases for a given block
+	powQuitChan := make(chan ethreact.Event, 1) // This is the channel that can exit the miner thread
 	quitChan := make(chan bool, 1)
 
 	ethereum.Reactor().Subscribe("newBlock", reactChan)