test event bus

Refs #693

libs/pubsub/query/Makefile

@@ -1,10 +1,10 @@
 gen_query_parser:
-	@go get github.com/pointlander/peg
+	go get -u -v github.com/pointlander/peg
 	peg -inline -switch query.peg
 
 fuzzy_test:
-	@go get github.com/dvyukov/go-fuzz/go-fuzz
-	@go get github.com/dvyukov/go-fuzz/go-fuzz-build
+	go get -u -v github.com/dvyukov/go-fuzz/go-fuzz
+	go get -u -v github.com/dvyukov/go-fuzz/go-fuzz-build
 	go-fuzz-build github.com/tendermint/tendermint/libs/pubsub/query/fuzz_test
 	go-fuzz -bin=./fuzz_test-fuzz.zip -workdir=./fuzz_test/output
 

libs/pubsub/query/query.peg.go

@@ -1,6 +1,7 @@
-// nolint
 package query
 
+//go:generate peg -inline -switch query.peg
+
 import (
 	"fmt"
 	"math"

types/event_bus_test.go

@@ -11,9 +11,9 @@ import (
 	"github.com/stretchr/testify/require"
 
 	abci "github.com/tendermint/tendermint/abci/types"
+	cmn "github.com/tendermint/tendermint/libs/common"
 	tmpubsub "github.com/tendermint/tendermint/libs/pubsub"
 	tmquery "github.com/tendermint/tendermint/libs/pubsub/query"
-	cmn "github.com/tendermint/tendermint/libs/common"
 )
 
 func TestEventBusPublishEventTx(t *testing.T) {
@@ -59,6 +59,64 @@ func TestEventBusPublishEventTx(t *testing.T) {
 	}
 }
 
+func TestEventBusPublish(t *testing.T) {
+	eventBus := NewEventBus()
+	err := eventBus.Start()
+	require.NoError(t, err)
+	defer eventBus.Stop()
+
+	eventsCh := make(chan interface{})
+	err = eventBus.Subscribe(context.Background(), "test", tmquery.Empty{}, eventsCh)
+	require.NoError(t, err)
+
+	const numEventsExpected = 14
+	done := make(chan struct{})
+	go func() {
+		numEvents := 0
+		for range eventsCh {
+			numEvents++
+			if numEvents >= numEventsExpected {
+				close(done)
+			}
+		}
+	}()
+
+	err = eventBus.Publish(EventNewBlockHeader, EventDataNewBlockHeader{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventNewBlock(EventDataNewBlock{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventNewBlockHeader(EventDataNewBlockHeader{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventVote(EventDataVote{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventProposalHeartbeat(EventDataProposalHeartbeat{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventNewRoundStep(EventDataRoundState{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventTimeoutPropose(EventDataRoundState{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventTimeoutWait(EventDataRoundState{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventNewRound(EventDataRoundState{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventCompleteProposal(EventDataRoundState{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventPolka(EventDataRoundState{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventUnlock(EventDataRoundState{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventRelock(EventDataRoundState{})
+	require.NoError(t, err)
+	err = eventBus.PublishEventLock(EventDataRoundState{})
+	require.NoError(t, err)
+
+	select {
+	case <-done:
+	case <-time.After(1 * time.Second):
+		t.Fatalf("expected to receive %d events after 1 sec.", numEventsExpected)
+	}
+}
+
 func BenchmarkEventBus(b *testing.B) {
 	benchmarks := []struct {
 		name        string