mqtt transport v2

Signed-off-by: Wei Liu <[email protected]>

Author: skeeey

pkg/cloudevents/generic/options/v2/mqtt/transport.go

@@ -58,7 +58,11 @@ func (t *mqttTransport) Connect(ctx context.Context) error {
 		ClientID: t.clientID,
 		Conn:     tcpConn,
 		OnClientError: func(err error) {
-			t.errorChan <- err
+			select {
+			case t.errorChan <- err:
+			default:
+				logger.Error(err, "mqtt client error")
+			}
 		},
 	}
 
@@ -199,29 +203,18 @@ func (t *mqttTransport) Close(ctx context.Context) error {
 	}
 
 	// Guard against double-close panic - check if already closed
-	alreadyClosed := false
 	if t.closeChan != nil {
 		select {
 		case <-t.closeChan:
-			// Already closed
-			alreadyClosed = true
+			// already closed
 		default:
 			close(t.closeChan)
 		}
 	}
 
-	// Only close msgChan if not already closed
-	if !alreadyClosed && t.msgChan != nil {
-		close(t.msgChan)
-	}
-
 	t.subscribed = false
 
-	if !alreadyClosed {
-		return t.client.Disconnect(&paho.Disconnect{ReasonCode: 0})
-	}
-
-	return nil
+	return t.client.Disconnect(&paho.Disconnect{ReasonCode: 0})
 }
 
 func (t *mqttTransport) getCloseChan() chan struct{} {

pkg/cloudevents/generic/options/v2/mqtt/transport_test.go

@@ -85,7 +85,7 @@ func setupTestBroker(t *testing.T) (*mochimqtt.Server, func()) {
 }
 
 // createTestMQTTOptions creates MQTTOptions for testing
-func createTestMQTTOptions(topic string) *mqtt.MQTTOptions {
+func createTestMQTTOptions() *mqtt.MQTTOptions {
 	return &mqtt.MQTTOptions{
 		KeepAlive: 60,
 		PubQoS:    1,
@@ -99,7 +99,7 @@ func createTestMQTTOptions(topic string) *mqtt.MQTTOptions {
 
 // createTestTransport creates a transport for testing
 func createTestTransport(clientID, pubTopic, subTopic string) *mqttTransport {
-	opts := createTestMQTTOptions(pubTopic)
+	opts := createTestMQTTOptions()
 	return newTransport(
 		clientID,
 		opts,
@@ -206,7 +206,9 @@ func TestTransportSend(t *testing.T) {
 	event.SetID("test-event-1")
 	event.SetSource("test-source")
 	event.SetType("test.type")
-	event.SetData(cloudevents.ApplicationJSON, map[string]string{"key": "value"})
+	if err := event.SetData(cloudevents.ApplicationJSON, map[string]string{"key": "value"}); err != nil {
+		t.Fatalf("failed to set data: %v", err)
+	}
 
 	// Send the event
 	if err := transport.Send(ctx, event); err != nil {
@@ -290,7 +292,9 @@ func TestTransportSubscribeAndReceive(t *testing.T) {
 	event.SetID("test-event-123")
 	event.SetSource("test-source")
 	event.SetType("test.type.pubsub")
-	event.SetData(cloudevents.ApplicationJSON, map[string]string{"message": "hello"})
+	if err := event.SetData(cloudevents.ApplicationJSON, map[string]string{"message": "hello"}); err != nil {
+		t.Fatalf("failed to set data: %v", err)
+	}
 
 	if err := sender.Send(ctx, event); err != nil {
 		t.Fatalf("failed to send: %v", err)
@@ -369,7 +373,9 @@ func TestTransportNoMessageLoss(t *testing.T) {
 		event.SetID(fmt.Sprintf("event-%d", i))
 		event.SetSource("test-source")
 		event.SetType("test.type")
-		event.SetData(cloudevents.ApplicationJSON, map[string]int{"seq": i})
+		if err := event.SetData(cloudevents.ApplicationJSON, map[string]int{"seq": i}); err != nil {
+			t.Fatalf("failed to set data: %v", err)
+		}
 
 		if err := sender.Send(ctx, event); err != nil {
 			t.Fatalf("failed to send event %d: %v", i, err)