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.

253 lines
7.2 KiB

  1. package pubsub_test
  2. import (
  3. "context"
  4. "fmt"
  5. "runtime/debug"
  6. "testing"
  7. "time"
  8. "github.com/stretchr/testify/assert"
  9. "github.com/stretchr/testify/require"
  10. "github.com/tendermint/tendermint/libs/log"
  11. "github.com/tendermint/tendermint/libs/pubsub"
  12. "github.com/tendermint/tendermint/libs/pubsub/query"
  13. )
  14. const (
  15. clientID = "test-client"
  16. )
  17. func TestSubscribe(t *testing.T) {
  18. s := pubsub.NewServer()
  19. s.SetLogger(log.TestingLogger())
  20. s.Start()
  21. defer s.Stop()
  22. ctx := context.Background()
  23. ch := make(chan interface{}, 1)
  24. err := s.Subscribe(ctx, clientID, query.Empty{}, ch)
  25. require.NoError(t, err)
  26. err = s.Publish(ctx, "Ka-Zar")
  27. require.NoError(t, err)
  28. assertReceive(t, "Ka-Zar", ch)
  29. err = s.Publish(ctx, "Quicksilver")
  30. require.NoError(t, err)
  31. assertReceive(t, "Quicksilver", ch)
  32. }
  33. func TestDifferentClients(t *testing.T) {
  34. s := pubsub.NewServer()
  35. s.SetLogger(log.TestingLogger())
  36. s.Start()
  37. defer s.Stop()
  38. ctx := context.Background()
  39. ch1 := make(chan interface{}, 1)
  40. err := s.Subscribe(ctx, "client-1", query.MustParse("tm.events.type='NewBlock'"), ch1)
  41. require.NoError(t, err)
  42. err = s.PublishWithTags(ctx, "Iceman", pubsub.NewTagMap(map[string]string{"tm.events.type": "NewBlock"}))
  43. require.NoError(t, err)
  44. assertReceive(t, "Iceman", ch1)
  45. ch2 := make(chan interface{}, 1)
  46. err = s.Subscribe(ctx, "client-2", query.MustParse("tm.events.type='NewBlock' AND abci.account.name='Igor'"), ch2)
  47. require.NoError(t, err)
  48. err = s.PublishWithTags(ctx, "Ultimo", pubsub.NewTagMap(map[string]string{"tm.events.type": "NewBlock", "abci.account.name": "Igor"}))
  49. require.NoError(t, err)
  50. assertReceive(t, "Ultimo", ch1)
  51. assertReceive(t, "Ultimo", ch2)
  52. ch3 := make(chan interface{}, 1)
  53. err = s.Subscribe(ctx, "client-3", query.MustParse("tm.events.type='NewRoundStep' AND abci.account.name='Igor' AND abci.invoice.number = 10"), ch3)
  54. require.NoError(t, err)
  55. err = s.PublishWithTags(ctx, "Valeria Richards", pubsub.NewTagMap(map[string]string{"tm.events.type": "NewRoundStep"}))
  56. require.NoError(t, err)
  57. assert.Zero(t, len(ch3))
  58. }
  59. func TestClientSubscribesTwice(t *testing.T) {
  60. s := pubsub.NewServer()
  61. s.SetLogger(log.TestingLogger())
  62. s.Start()
  63. defer s.Stop()
  64. ctx := context.Background()
  65. q := query.MustParse("tm.events.type='NewBlock'")
  66. ch1 := make(chan interface{}, 1)
  67. err := s.Subscribe(ctx, clientID, q, ch1)
  68. require.NoError(t, err)
  69. err = s.PublishWithTags(ctx, "Goblin Queen", pubsub.NewTagMap(map[string]string{"tm.events.type": "NewBlock"}))
  70. require.NoError(t, err)
  71. assertReceive(t, "Goblin Queen", ch1)
  72. ch2 := make(chan interface{}, 1)
  73. err = s.Subscribe(ctx, clientID, q, ch2)
  74. require.Error(t, err)
  75. err = s.PublishWithTags(ctx, "Spider-Man", pubsub.NewTagMap(map[string]string{"tm.events.type": "NewBlock"}))
  76. require.NoError(t, err)
  77. assertReceive(t, "Spider-Man", ch1)
  78. }
  79. func TestUnsubscribe(t *testing.T) {
  80. s := pubsub.NewServer()
  81. s.SetLogger(log.TestingLogger())
  82. s.Start()
  83. defer s.Stop()
  84. ctx := context.Background()
  85. ch := make(chan interface{})
  86. err := s.Subscribe(ctx, clientID, query.MustParse("tm.events.type='NewBlock'"), ch)
  87. require.NoError(t, err)
  88. err = s.Unsubscribe(ctx, clientID, query.MustParse("tm.events.type='NewBlock'"))
  89. require.NoError(t, err)
  90. err = s.Publish(ctx, "Nick Fury")
  91. require.NoError(t, err)
  92. assert.Zero(t, len(ch), "Should not receive anything after Unsubscribe")
  93. _, ok := <-ch
  94. assert.False(t, ok)
  95. }
  96. func TestResubscribe(t *testing.T) {
  97. s := pubsub.NewServer()
  98. s.SetLogger(log.TestingLogger())
  99. s.Start()
  100. defer s.Stop()
  101. ctx := context.Background()
  102. ch := make(chan interface{})
  103. err := s.Subscribe(ctx, clientID, query.Empty{}, ch)
  104. require.NoError(t, err)
  105. err = s.Unsubscribe(ctx, clientID, query.Empty{})
  106. require.NoError(t, err)
  107. ch = make(chan interface{})
  108. err = s.Subscribe(ctx, clientID, query.Empty{}, ch)
  109. require.NoError(t, err)
  110. err = s.Publish(ctx, "Cable")
  111. require.NoError(t, err)
  112. assertReceive(t, "Cable", ch)
  113. }
  114. func TestUnsubscribeAll(t *testing.T) {
  115. s := pubsub.NewServer()
  116. s.SetLogger(log.TestingLogger())
  117. s.Start()
  118. defer s.Stop()
  119. ctx := context.Background()
  120. ch1, ch2 := make(chan interface{}, 1), make(chan interface{}, 1)
  121. err := s.Subscribe(ctx, clientID, query.MustParse("tm.events.type='NewBlock'"), ch1)
  122. require.NoError(t, err)
  123. err = s.Subscribe(ctx, clientID, query.MustParse("tm.events.type='NewBlockHeader'"), ch2)
  124. require.NoError(t, err)
  125. err = s.UnsubscribeAll(ctx, clientID)
  126. require.NoError(t, err)
  127. err = s.Publish(ctx, "Nick Fury")
  128. require.NoError(t, err)
  129. assert.Zero(t, len(ch1), "Should not receive anything after UnsubscribeAll")
  130. assert.Zero(t, len(ch2), "Should not receive anything after UnsubscribeAll")
  131. _, ok := <-ch1
  132. assert.False(t, ok)
  133. _, ok = <-ch2
  134. assert.False(t, ok)
  135. }
  136. func TestBufferCapacity(t *testing.T) {
  137. s := pubsub.NewServer(pubsub.BufferCapacity(2))
  138. s.SetLogger(log.TestingLogger())
  139. assert.Equal(t, 2, s.BufferCapacity())
  140. ctx := context.Background()
  141. err := s.Publish(ctx, "Nighthawk")
  142. require.NoError(t, err)
  143. err = s.Publish(ctx, "Sage")
  144. require.NoError(t, err)
  145. ctx, cancel := context.WithTimeout(ctx, 10*time.Millisecond)
  146. defer cancel()
  147. err = s.Publish(ctx, "Ironclad")
  148. if assert.Error(t, err) {
  149. assert.Equal(t, context.DeadlineExceeded, err)
  150. }
  151. }
  152. func Benchmark10Clients(b *testing.B) { benchmarkNClients(10, b) }
  153. func Benchmark100Clients(b *testing.B) { benchmarkNClients(100, b) }
  154. func Benchmark1000Clients(b *testing.B) { benchmarkNClients(1000, b) }
  155. func Benchmark10ClientsOneQuery(b *testing.B) { benchmarkNClientsOneQuery(10, b) }
  156. func Benchmark100ClientsOneQuery(b *testing.B) { benchmarkNClientsOneQuery(100, b) }
  157. func Benchmark1000ClientsOneQuery(b *testing.B) { benchmarkNClientsOneQuery(1000, b) }
  158. func benchmarkNClients(n int, b *testing.B) {
  159. s := pubsub.NewServer()
  160. s.Start()
  161. defer s.Stop()
  162. ctx := context.Background()
  163. for i := 0; i < n; i++ {
  164. ch := make(chan interface{})
  165. go func() {
  166. for range ch {
  167. }
  168. }()
  169. s.Subscribe(ctx, clientID, query.MustParse(fmt.Sprintf("abci.Account.Owner = 'Ivan' AND abci.Invoices.Number = %d", i)), ch)
  170. }
  171. b.ReportAllocs()
  172. b.ResetTimer()
  173. for i := 0; i < b.N; i++ {
  174. s.PublishWithTags(ctx, "Gamora", pubsub.NewTagMap(map[string]string{"abci.Account.Owner": "Ivan", "abci.Invoices.Number": string(i)}))
  175. }
  176. }
  177. func benchmarkNClientsOneQuery(n int, b *testing.B) {
  178. s := pubsub.NewServer()
  179. s.Start()
  180. defer s.Stop()
  181. ctx := context.Background()
  182. q := query.MustParse("abci.Account.Owner = 'Ivan' AND abci.Invoices.Number = 1")
  183. for i := 0; i < n; i++ {
  184. ch := make(chan interface{})
  185. go func() {
  186. for range ch {
  187. }
  188. }()
  189. s.Subscribe(ctx, clientID, q, ch)
  190. }
  191. b.ReportAllocs()
  192. b.ResetTimer()
  193. for i := 0; i < b.N; i++ {
  194. s.PublishWithTags(ctx, "Gamora", pubsub.NewTagMap(map[string]string{"abci.Account.Owner": "Ivan", "abci.Invoices.Number": "1"}))
  195. }
  196. }
  197. ///////////////////////////////////////////////////////////////////////////////
  198. /// HELPERS
  199. ///////////////////////////////////////////////////////////////////////////////
  200. func assertReceive(t *testing.T, expected interface{}, ch <-chan interface{}, msgAndArgs ...interface{}) {
  201. select {
  202. case actual := <-ch:
  203. if actual != nil {
  204. assert.Equal(t, expected, actual, msgAndArgs...)
  205. }
  206. case <-time.After(1 * time.Second):
  207. t.Errorf("Expected to receive %v from the channel, got nothing after 1s", expected)
  208. debug.PrintStack()
  209. }
  210. }