| 74 | } |
| 75 | |
| 76 | func TestMultiTopic(t *testing.T) { |
| 77 | q := New("channel") |
| 78 | |
| 79 | //mempool |
| 80 | go func() { |
| 81 | client := q.Client() |
| 82 | client.Sub("mempool") |
| 83 | for msg := range client.Recv() { |
| 84 | if msg.Ty == types.EventTx { |
| 85 | msg.Reply(client.NewMessage("mempool", types.EventReply, types.Reply{IsOk: true, Msg: []byte("word")})) |
| 86 | } |
| 87 | } |
| 88 | }() |
| 89 | |
| 90 | //blockchain |
| 91 | go func() { |
| 92 | client := q.Client() |
| 93 | client.Sub("blockchain") |
| 94 | for msg := range client.Recv() { |
| 95 | if msg.Ty == types.EventGetBlockHeight { |
| 96 | msg.Reply(client.NewMessage("blockchain", types.EventReplyBlockHeight, types.ReplyBlockHeight{Height: 100})) |
| 97 | } |
| 98 | } |
| 99 | }() |
| 100 | |
| 101 | //rpc server |
| 102 | go func() { |
| 103 | client := q.Client() |
| 104 | //rpc 模块 会向其他模块发送消息，自己本身不需要订阅消息 |
| 105 | msg := client.NewMessage("mempool", types.EventTx, "hello") |
| 106 | client.Send(msg, true) |
| 107 | reply, err := client.Wait(msg) |
| 108 | if err != nil { |
| 109 | t.Error(err) |
| 110 | return |
| 111 | } |
| 112 | t.Log(string(reply.GetData().(types.Reply).Msg)) |
| 113 | |
| 114 | msg = client.NewMessage("blockchain", types.EventGetBlockHeight, nil) |
| 115 | client.Send(msg, true) |
| 116 | reply, err = client.Wait(msg) |
| 117 | if err != nil { |
| 118 | t.Error(err) |
| 119 | return |
| 120 | } |
| 121 | t.Log(reply) |
| 122 | q.Close() |
| 123 | }() |
| 124 | q.Start() |
| 125 | } |
| 126 | |
| 127 | //发送100000 低优先级的消息，然后发送一个高优先级的消息 |
| 128 | //高优先级的消息可以即时返回 |