diff --git a/internal/consensus/iterator.go b/internal/consensus/iterator.go index 5680c0c18..97b962283 100644 --- a/internal/consensus/iterator.go +++ b/internal/consensus/iterator.go @@ -2,23 +2,10 @@ package consensus import ( "context" - "errors" - "fmt" "io" "time" ) -const ( - Timeout EventType = iota - ReceiveMessage -) - -var ( - ErrEndOfIterator = errors.New("end of iterator") -) - -type EventType int64 - type txAvailable struct{} type Event struct { @@ -39,7 +26,6 @@ type WALReadingIterator struct { } func (w WALReadingIterator) Next(ctx context.Context) bool { - fmt.Println("hello") wm, err := w.dec.Decode() if err != nil { if err != io.EOF { @@ -78,6 +64,42 @@ func (t testIterator) Next(ctx context.Context) bool { func (t testIterator) Event() Event { return t.events[t.i] } + func (t testIterator) Error() error { return t.e } + +type LiveIterator struct { + peerMsgQueue chan msgInfo + internalMsgQueue chan msgInfo + timeoutTicker TimeoutTicker + txNotifier txNotifier + + err error + e Event +} + +func (l *LiveIterator) Next(ctx context.Context) bool { + select { + case m := <-l.peerMsgQueue: + l.e = Event{m, time.Now()} + case m := <-l.internalMsgQueue: + l.e = Event{m, time.Now()} + case t := <-l.timeoutTicker.Chan(): + l.e = Event{Message: t, Time: time.Now()} + case <-l.txNotifier.TxsAvailable(): + l.e = Event{txAvailable{}, time.Now()} + case <-ctx.Done(): + l.err = ctx.Err() + return false + } + return true +} + +func (l *LiveIterator) Event() Event { + return l.e +} + +func (l *LiveIterator) Error() error { + return l.err +} diff --git a/internal/consensus/state.go b/internal/consensus/state.go index 715589860..fad9648a6 100644 --- a/internal/consensus/state.go +++ b/internal/consensus/state.go @@ -442,7 +442,14 @@ func (cs *State) OnStart(ctx context.Context) error { } // now start the receiveRoutine - go cs.receiveRoutine(ctx, testIterator{}, 0) + iter := LiveIterator{ + peerMsgQueue: cs.peerMsgQueue, + internalMsgQueue: cs.internalMsgQueue, + timeoutTicker: cs.timeoutTicker, + txNotifier: cs.txNotifier, + } + + go cs.receiveRoutine(ctx, &iter, 0) // schedule the first round! // use GetRoundState so we don't race the receiveRoutine for access