Function TestHandlerDownsideMqtt

engine/msg_handler_test.go:538–599

(t *testing.T)

}

func TestHandlerDownsideMqtt(t *testing.T) {
	e, _, _ := genHandlerDownsideEngine(t)
	h := &handlerDownside{engineImpl: e}

	// upside msg
	ch, err := e.pb.Subscribe(sync.TopicUpside)
	assert.NoError(t, err)
	assert.NotNil(t, ch)

	handler := &msgRPCUpside{t: t}
	pro := pubsub.NewProcessor(ch, 0, handler)
	pro.Start()
	defer pro.Close()

	// event msg
	ch0, err := e.pb.Subscribe(eventx.TopicEvent)
	assert.NoError(t, err)
	assert.NotNil(t, ch)

	handler0 := &msgRPCMqtt{t: t}
	pro0 := pubsub.NewProcessor(ch0, 0, handler0)
	pro0.Start()
	defer pro0.Close()

	// req0 rpc success
	engMsgWG.Add(2)
	req0 := &specV1.Message{
		Kind: specV1.MessageCMD,
		Metadata: map[string]string{
			"cmd": specV1.MessageRPCMqtt,
		},
		Content: specV1.LazyValue{
			Value: &specV1.RPCMqttMessage{
				Topic:   "test/node",
				QoS:     0,
				Content: "result",
			},
		},
	}
	handler.check = func(msg interface{}) {
		m, ok := msg.(*specV1.Message)
		assert.True(t, ok)
		assert.Equal(t, "true", m.Metadata["success"])
		engMsgWG.Done()
	}
	handler0.check = func(msg interface{}) {
		m, ok := msg.(*specV1.Message)
		assert.True(t, ok)
		request := &specV1.RPCMqttMessage{}
		handlerErr := m.Content.Unmarshal(request)
		assert.NoError(t, handlerErr)
		var buf []byte
		if request.Content != nil {
			buf = []byte(fmt.Sprintf("%v", request.Content))
		}
		assert.Equal(t, string(buf), "result")
		engMsgWG.Done()
	}