Simplify Service/Reactor pattern

blockchain/pool.go

@@ -68,12 +68,14 @@ func NewBlockPool(start int, requestsCh chan<- BlockRequest, timeoutsCh chan<- s
 	return bp
 }
 
-func (pool *BlockPool) AfterStart() {
+func (pool *BlockPool) OnStart() {
+	pool.BaseService.OnStart()
 	pool.repeater = NewRepeatTimer("", requestIntervalMS*time.Millisecond)
 	go pool.run()
 }
 
-func (pool *BlockPool) AfterStop() {
+func (pool *BlockPool) OnStop() {
+	pool.BaseService.OnStop()
 	pool.repeater.Stop()
 }
 

blockchain/reactor.go

@@ -76,14 +76,16 @@ func NewBlockchainReactor(state *sm.State, store *BlockStore, sync bool) *Blockc
 	return bcR
 }
 
-func (bcR *BlockchainReactor) AfterStart() {
+func (bcR *BlockchainReactor) OnStart() {
+	bcR.BaseReactor.OnStart()
 	if bcR.sync {
 		bcR.pool.Start()
 		go bcR.poolRoutine()
 	}
 }
 
-func (bcR *BlockchainReactor) AfterStop() {
+func (bcR *BlockchainReactor) OnStop() {
+	bcR.BaseReactor.OnStop()
 	bcR.pool.Stop()
 }
 

common/service.go

@@ -2,7 +2,7 @@
 
 Classical-inheritance-style service declarations.
 Services can be started, then stopped.
-Users can override the AfterStart/AfterStop methods.
+Users can override the OnStart/OnStop methods.
 These methods are guaranteed to be called at most once.
 Caller must ensure that Start() and Stop() are not called concurrently.
 It is ok to call Stop() without calling Start() first.
@@ -19,16 +19,18 @@ func NewFooService() *FooService {
 	fs := &FooService{
 		// init
 	}
-	fs.BaseService = *BaseService(log, "FooService", fs)
+	fs.BaseService = *NewBaseService(log, "FooService", fs)
 	return fs
 }
 
-func (fs *FooService) AfterStart() {
+func (fs *FooService) OnStart() {
+	fs.BaseService.OnStart() // Always call the overridden method.
 	// initialize private fields
 	// start subroutines, etc.
 }
 
-func (fs *FooService) AfterStart() {
+func (fs *FooService) OnStop() {
+	fs.BaseService.OnStop() // Always call the overridden method.
 	// close/destroy private fields
 	// stop subroutines, etc.
 }
@@ -41,12 +43,10 @@ import "github.com/tendermint/tendermint/Godeps/_workspace/src/github.com/tender
 
 type Service interface {
 	Start() bool
-	BeforeStart()
+	OnStart()
-	AfterStart()
 
 	Stop() bool
-	BeforeStop()
+	OnStop()
-	AfterStop()
 
 	IsRunning() bool
 
@@ -80,8 +80,7 @@ func (bs *BaseService) Start() bool {
 		} else {
 			bs.log.Notice(Fmt("Starting %v", bs.name), "impl", bs.impl)
 		}
-		bs.impl.BeforeStart()
+		bs.impl.OnStart()
-		bs.impl.AfterStart()
 		return true
 	} else {
 		bs.log.Info(Fmt("Not starting %v -- already started", bs.name), "impl", bs.impl)
@@ -90,17 +89,13 @@ func (bs *BaseService) Start() bool {
 }
 
 // Implements Service
-func (bs *BaseService) BeforeStart() {}
+func (bs *BaseService) OnStart() {}
-
-// Implements Service
-func (bs *BaseService) AfterStart() {}
 
 // Implements Service
 func (bs *BaseService) Stop() bool {
 	if atomic.CompareAndSwapUint32(&bs.stopped, 0, 1) {
 		bs.log.Notice(Fmt("Stopping %v", bs.name), "impl", bs.impl)
-		bs.impl.BeforeStop()
+		bs.impl.OnStop()
-		bs.impl.AfterStop()
 		return true
 	} else {
 		bs.log.Notice(Fmt("Not stopping %v", bs.name), "impl", bs.impl)
@@ -109,10 +104,7 @@ func (bs *BaseService) Stop() bool {
 }
 
 // Implements Service
-func (bs *BaseService) BeforeStop() {}
+func (bs *BaseService) OnStop() {}
-
-// Implements Service
-func (bs *BaseService) AfterStop() {}
 
 // Implements Service
 func (bs *BaseService) IsRunning() bool {
@@ -138,17 +130,12 @@ func NewQuitService(log log15.Logger, name string, impl Service) *QuitService {
 	}
 }
 
-// Init .Quit in BeforeStart such that AfterStart of impls have access to Quit.
+// NOTE: when overriding OnStart, must call .QuitService.OnStart().
-// NOTE: When overriding BeforeStart, call QuitService.BeforeStart() manually.
+func (qs *QuitService) OnStart() {
-func (qs *QuitService) BeforeStart() {
 	qs.Quit = make(chan struct{})
 }
 
-// Close .Quit after Stop/BeforeStop/AfterStop
+// NOTE: when overriding OnStop, must call .QuitService.OnStop().
-func (qs *QuitService) Stop() bool {
+func (qs *QuitService) OnStop() {
-	res := qs.BaseService.Stop()
+	close(qs.Quit)
-	if res {
-		close(qs.Quit)
-	}
-	return res
 }

consensus/reactor.go

@@ -49,15 +49,17 @@ func NewConsensusReactor(consensusState *ConsensusState, blockStore *bc.BlockSto
 	return conR
 }
 
-func (conR *ConsensusReactor) AfterStart() {
+func (conR *ConsensusReactor) OnStart() {
 	log.Notice("ConsensusReactor ", "fastSync", conR.fastSync)
+	conR.BaseReactor.OnStart()
 	if !conR.fastSync {
 		conR.conS.Start()
 	}
 	go conR.broadcastNewRoundStepRoutine()
 }
 
-func (conR *ConsensusReactor) AfterStop() {
+func (conR *ConsensusReactor) OnStop() {
+	conR.BaseReactor.OnStop()
 	conR.conS.Stop()
 }
 

consensus/state.go

@@ -360,12 +360,14 @@ func (cs *ConsensusState) NewStepCh() chan *RoundState {
 	return cs.newStepCh
 }
 
-func (cs *ConsensusState) AfterStart() {
+func (cs *ConsensusState) OnStart() {
+	cs.BaseService.OnStart()
 	cs.scheduleRound0(cs.Height)
 }
 
-func (cs *ConsensusState) AfterStop() {
+func (cs *ConsensusState) OnStop() {
 	// It's mostly asynchronous so, there's not much to stop.
+	cs.BaseService.OnStop()
 }
 
 // EnterNewRound(height, 0) at cs.StartTime.

events/events.go

@@ -31,12 +31,14 @@ func NewEventSwitch() *EventSwitch {
 	return evsw
 }
 
-func (evsw *EventSwitch) AfterStart() {
+func (evsw *EventSwitch) OnStart() {
+	evsw.BaseService.OnStart()
 	evsw.eventCells = make(map[string]*eventCell)
 	evsw.listeners = make(map[string]*eventListener)
 }
 
-func (evsw *EventSwitch) AfterStop() {
+func (evsw *EventSwitch) OnStop() {
+	evsw.BaseService.OnStop()
 	evsw.eventCells = nil
 	evsw.listeners = nil
 }

mempool/reactor.go

@@ -34,9 +34,9 @@ func NewMempoolReactor(mempool *Mempool) *MempoolReactor {
 	return memR
 }
 
-// func (memR *MempoolReactor) AfterStart() {}
+// func (memR *MempoolReactor) OnStart() { memR.BaseReactor.OnStart() }
 
-// func (memR *MempoolReactor) AfterStop() {}
+// func (memR *MempoolReactor) OnStop() { memR.BaseReactor.OnStop() }
 
 // Implements Reactor
 func (memR *MempoolReactor) GetChannels() []*p2p.ChannelDescriptor {

p2p/addrbook.go

@@ -121,13 +121,15 @@ func (a *AddrBook) init() {
 	}
 }
 
-func (a *AddrBook) AfterStart() {
+func (a *AddrBook) OnStart() {
+	a.QuitService.OnStart()
 	a.loadFromFile(a.filePath)
 	a.wg.Add(1)
 	go a.saveRoutine()
 }
 
-func (a *AddrBook) AfterStop() {
+func (a *AddrBook) OnStop() {
+	a.QuitService.OnStop()
 	a.wg.Wait()
 }
 

p2p/connection.go

@@ -127,7 +127,8 @@ func NewMConnection(conn net.Conn, chDescs []*ChannelDescriptor, onReceive recei
 	return mconn
 }
 
-func (c *MConnection) AfterStart() {
+func (c *MConnection) OnStart() {
+	c.BaseService.OnStart()
 	c.quit = make(chan struct{})
 	c.flushTimer = NewThrottleTimer("flush", flushThrottleMS*time.Millisecond)
 	c.pingTimer = NewRepeatTimer("ping", pingTimeoutSeconds*time.Second)
@@ -136,7 +137,8 @@ func (c *MConnection) AfterStart() {
 	go c.recvRoutine()
 }
 
-func (c *MConnection) AfterStop() {
+func (c *MConnection) OnStop() {
+	c.BaseService.OnStop()
 	c.flushTimer.Stop()
 	c.pingTimer.Stop()
 	c.chStatsTimer.Stop()

p2p/listener.go

@@ -97,11 +97,13 @@ SKIP_UPNP:
 	return dl
 }
 
-func (l *DefaultListener) AfterStart() {
+func (l *DefaultListener) OnStart() {
+	l.BaseService.OnStart()
 	go l.listenRoutine()
 }
 
-func (l *DefaultListener) AfterStop() {
+func (l *DefaultListener) OnStop() {
+	l.BaseService.OnStop()
 	l.listener.Close()
 }
 

p2p/peer.go

@@ -72,11 +72,13 @@ func newPeer(conn net.Conn, peerNodeInfo *types.NodeInfo, outbound bool, reactor
 	return p
 }
 
-func (p *Peer) AfterStart() {
+func (p *Peer) OnStart() {
+	p.BaseService.OnStart()
 	p.mconn.Start()
 }
 
-func (p *Peer) AfterStop() {
+func (p *Peer) OnStop() {
+	p.BaseService.OnStop()
 	p.mconn.Stop()
 }
 

p2p/pex_reactor.go

@@ -41,11 +41,13 @@ func NewPEXReactor(book *AddrBook) *PEXReactor {
 	return pexR
 }
 
-func (pexR *PEXReactor) AfterStart() {
+func (pexR *PEXReactor) OnStart() {
+	pexR.BaseReactor.OnStart()
 	go pexR.ensurePeersRoutine()
 }
 
-func (pexR *PEXReactor) AfterStop() {
+func (pexR *PEXReactor) OnStop() {
+	pexR.BaseReactor.OnStop()
 }
 
 // Implements Reactor

p2p/switch.go

@@ -153,7 +153,8 @@ func (sw *Switch) SetNodePrivKey(nodePrivKey acm.PrivKeyEd25519) {
 }
 
 // Switch.Start() starts all the reactors, peers, and listeners.
-func (sw *Switch) AfterStart() {
+func (sw *Switch) OnStart() {
+	sw.BaseService.OnStart()
 	// Start reactors
 	for _, reactor := range sw.reactors {
 		reactor.Start()
@@ -168,7 +169,8 @@ func (sw *Switch) AfterStart() {
 	}
 }
 
-func (sw *Switch) AfterStop() {
+func (sw *Switch) OnStop() {
+	sw.BaseService.OnStop()
 	// Stop listeners
 	for _, listener := range sw.listeners {
 		listener.Stop()

rpc/server/handlers.go

@@ -241,14 +241,16 @@ func NewWSConnection(wsConn *websocket.Conn) *WSConnection {
 	return con
 }
 
-func (con *WSConnection) AfterStart() {
+func (con *WSConnection) OnStart() {
+	con.QuitService.OnStart()
 	// read subscriptions/unsubscriptions to events
 	go con.read()
 	// write responses
 	con.write()
 }
 
-func (con *WSConnection) AfterStop() {
+func (con *WSConnection) OnStop() {
+	con.QuitService.OnStop()
 	con.evsw.RemoveListener(con.id)
 	// the write loop closes the websocket connection
 	// when it exits its loop, and the read loop