+
+func PeerTapProcessor(peer *Peer, tap *TAP, terminator chan struct{}) {
+ var data []byte
+ var now time.Time
+ lastSent := time.Now()
+ heartbeat := time.NewTicker(peer.Timeout)
+ if peer.CPRCycle == time.Duration(0) {
+ RawProcessor:
+ for {
+ select {
+ case <-terminator:
+ break RawProcessor
+ case <-heartbeat.C:
+ now = time.Now()
+ if lastSent.Add(peer.Timeout).Before(now) {
+ peer.EthProcess(nil)
+ lastSent = now
+ }
+ case data = <-tap.Sink:
+ peer.EthProcess(data)
+ lastSent = time.Now()
+ }
+ }
+ } else {
+ CPRProcessor:
+ for {
+ data = nil
+ select {
+ case <-terminator:
+ break CPRProcessor
+ case data = <-tap.Sink:
+ peer.EthProcess(data)
+ default:
+ }
+ if data == nil {
+ peer.EthProcess(nil)
+ }
+ time.Sleep(peer.CPRCycle)
+ }
+ }
+ close(terminator)
+ peer.Zero()
+ heartbeat.Stop()
+}