Add todos for fixing response waiters

messagix/events.go

@@ -279,6 +279,7 @@ func (pb *Event_PublishResponse) Finish() ResponseData {
 			Int("payload_length", len(pb.Data.Payload)).
 			Str("topic", pb.Topic).
 			Int64("request_id", pb.Data.RequestID).
+			Uint16("mqtt_message_id", pb.MessageIdentifier).
 			Strs("sp", pb.Data.Sp).
 			Int("target", pb.Data.Target)
 		if len(pb.Data.Payload) < 8192 {

messagix/responsehandler.go

@@ -70,6 +70,7 @@ func (p *ResponseHandler) waitForDetails(packetId uint16, channelType ChannelTyp
 		return response
 	case <-time.After(packetTimeout):
 		p.deleteDetails(packetId, channelType)
+		// TODO this is completely wrong, it should be a proper error
 		return &Event_PublishResponse{}
 	}
 }

messagix/socket.go

@@ -125,7 +125,7 @@ func (s *Socket) BuildBrokerUrl() string {
 }
 
 const pongTimeout = 30 * time.Second
-const packetTimeout = 10 * time.Second
+const packetTimeout = 30 * time.Second
 const pingInterval = 10 * time.Second
 
 func ptr[T any](val T) *T {
@@ -323,6 +323,7 @@ func (s *Socket) sendPublishPacket(topic Topic, jsonData string, packet *packets
 	}
 
 	s.responseHandler.addPacketChannel(packetId)
+	// TODO this should probably wait for the puback packet
 	return packetId, s.sendData(publishRequestPayload)
 }
 

messagix/threads.go

@@ -27,6 +27,8 @@ func (c *Client) ExecuteTasks(tasks ...socket.Task) (*table.LSTable, error) {
 	resp := c.socket.responseHandler.waitForPubResponseDetails(packetId)
 	if resp == nil {
 		return nil, fmt.Errorf("failed to receive response from socket after sending SendMessageTask. packetId: %d", packetId)
+	} else if resp.Topic == "" {
+		return nil, fmt.Errorf("request timed out")
 	}
 	resp.Finish()