You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

64 lines
1.7 KiB

pubsub 2.0 (#3227) * green pubsub tests :OK: * get rid of clientToQueryMap * Subscribe and SubscribeUnbuffered * start adapting other pkgs to new pubsub * nope * rename MsgAndTags to Message * remove TagMap it does not bring any additional benefits * bring back EventSubscriber * fix test * fix data race in TestStartNextHeightCorrectly ``` Write at 0x00c0001c7418 by goroutine 796: github.com/tendermint/tendermint/consensus.TestStartNextHeightCorrectly() /go/src/github.com/tendermint/tendermint/consensus/state_test.go:1296 +0xad testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 Previous read at 0x00c0001c7418 by goroutine 858: github.com/tendermint/tendermint/consensus.(*ConsensusState).addVote() /go/src/github.com/tendermint/tendermint/consensus/state.go:1631 +0x1366 github.com/tendermint/tendermint/consensus.(*ConsensusState).tryAddVote() /go/src/github.com/tendermint/tendermint/consensus/state.go:1476 +0x8f github.com/tendermint/tendermint/consensus.(*ConsensusState).handleMsg() /go/src/github.com/tendermint/tendermint/consensus/state.go:667 +0xa1e github.com/tendermint/tendermint/consensus.(*ConsensusState).receiveRoutine() /go/src/github.com/tendermint/tendermint/consensus/state.go:628 +0x794 Goroutine 796 (running) created at: testing.(*T).Run() /usr/local/go/src/testing/testing.go:878 +0x659 testing.runTests.func1() /usr/local/go/src/testing/testing.go:1119 +0xa8 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 testing.runTests() /usr/local/go/src/testing/testing.go:1117 +0x4ee testing.(*M).Run() /usr/local/go/src/testing/testing.go:1034 +0x2ee main.main() _testmain.go:214 +0x332 Goroutine 858 (running) created at: github.com/tendermint/tendermint/consensus.(*ConsensusState).startRoutines() /go/src/github.com/tendermint/tendermint/consensus/state.go:334 +0x221 github.com/tendermint/tendermint/consensus.startTestRound() /go/src/github.com/tendermint/tendermint/consensus/common_test.go:122 +0x63 github.com/tendermint/tendermint/consensus.TestStateFullRound1() /go/src/github.com/tendermint/tendermint/consensus/state_test.go:255 +0x397 testing.tRunner() /usr/local/go/src/testing/testing.go:827 +0x162 ``` * fixes after my own review * fix formatting * wait 100ms before kicking a subscriber out + a test for indexer_service * fixes after my second review * no timeout * add changelog entries * fix merge conflicts * fix typos after Thane's review Co-Authored-By: melekes <anton.kalyaev@gmail.com> * reformat code * rewrite indexer service in the attempt to fix failing test https://github.com/tendermint/tendermint/pull/3227/#issuecomment-462316527 * Revert "rewrite indexer service in the attempt to fix failing test" This reverts commit 0d9107a098230de7138abb1c201877c246e89ed1. * another attempt to fix indexer * fixes after Ethan's review * use unbuffered channel when indexing transactions Refs https://github.com/tendermint/tendermint/pull/3227#discussion_r258786716 * add a comment for EventBus#SubscribeUnbuffered * format code
6 years ago
  1. package txindex_test
  2. import (
  3. "testing"
  4. "time"
  5. "github.com/stretchr/testify/assert"
  6. "github.com/stretchr/testify/require"
  7. abci "github.com/tendermint/tendermint/abci/types"
  8. "github.com/tendermint/tendermint/libs/db"
  9. "github.com/tendermint/tendermint/libs/log"
  10. "github.com/tendermint/tendermint/state/txindex"
  11. "github.com/tendermint/tendermint/state/txindex/kv"
  12. "github.com/tendermint/tendermint/types"
  13. )
  14. func TestIndexerServiceIndexesBlocks(t *testing.T) {
  15. // event bus
  16. eventBus := types.NewEventBus()
  17. eventBus.SetLogger(log.TestingLogger())
  18. err := eventBus.Start()
  19. require.NoError(t, err)
  20. defer eventBus.Stop()
  21. // tx indexer
  22. store := db.NewMemDB()
  23. txIndexer := kv.NewTxIndex(store, kv.IndexAllTags())
  24. service := txindex.NewIndexerService(txIndexer, eventBus)
  25. service.SetLogger(log.TestingLogger())
  26. err = service.Start()
  27. require.NoError(t, err)
  28. defer service.Stop()
  29. // publish block with txs
  30. eventBus.PublishEventNewBlockHeader(types.EventDataNewBlockHeader{
  31. Header: types.Header{Height: 1, NumTxs: 2},
  32. })
  33. txResult1 := &types.TxResult{
  34. Height: 1,
  35. Index: uint32(0),
  36. Tx: types.Tx("foo"),
  37. Result: abci.ResponseDeliverTx{Code: 0},
  38. }
  39. eventBus.PublishEventTx(types.EventDataTx{*txResult1})
  40. txResult2 := &types.TxResult{
  41. Height: 1,
  42. Index: uint32(1),
  43. Tx: types.Tx("bar"),
  44. Result: abci.ResponseDeliverTx{Code: 0},
  45. }
  46. eventBus.PublishEventTx(types.EventDataTx{*txResult2})
  47. time.Sleep(100 * time.Millisecond)
  48. // check the result
  49. res, err := txIndexer.Get(types.Tx("foo").Hash())
  50. assert.NoError(t, err)
  51. assert.Equal(t, txResult1, res)
  52. res, err = txIndexer.Get(types.Tx("bar").Hash())
  53. assert.NoError(t, err)
  54. assert.Equal(t, txResult2, res)
  55. }