ranjith/MQTT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added logs for all errors

Browse Source
This commit is contained in:
ranjith
2025-07-28 06:18:26 +00:00
parent b1609bb4d4
commit 75e401c52f
1 changed files with 11 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
mqtt_client.go
View File

@@ -169,7 +169,7 @@ func messageHandler(client mqtt.Client, msg mqtt.Message) {
rows, err := db.Query(paramQuery, mfmMeterID, plantID) rows, err := db.Query(paramQuery, mfmMeterID, plantID)
if err != nil { if err != nil {
publishError(client, errorTopic, fmt.Sprintf("❌ Failed to fetch parameters for meter %d: %v", mfmMeterID, err)) logAndPublishError(client, errorTopic, fmt.Sprintf("❌ Failed to fetch parameters for meter %d: %v", mfmMeterID, err))
return return
} }
defer rows.Close() defer rows.Close()
@@ -178,25 +178,25 @@ func messageHandler(client mqtt.Client, msg mqtt.Message) {
for rows.Next() { for rows.Next() {
var name string var name string
if err := rows.Scan(&name); err != nil { if err := rows.Scan(&name); err != nil {
publishError(client, errorTopic, fmt.Sprintf("❌ Failed to scan parameter name: %v", err)) logAndPublishError(client, errorTopic, fmt.Sprintf("❌ Failed to scan parameter name: %v", err))
return return
} }
paramColumns = append(paramColumns, name) paramColumns = append(paramColumns, name)
} }
if len(paramColumns) == 0 { if len(paramColumns) == 0 {
publishError(client, errorTopic, fmt.Sprintf("❌ No parameters found for meter_id %d", mfmMeterID)) logAndPublishError(client, errorTopic, fmt.Sprintf("❌ No parameters found for meter_id %d", mfmMeterID))
return return
} }
// Validate values count // Validate values count
if len(values) < len(paramColumns) { if len(values) < len(paramColumns) {
publishError(client, errorTopic, fmt.Sprintf("❌ Insufficient register values: expected %d, got %d", len(paramColumns), len(values))) logAndPublishError(client, errorTopic, fmt.Sprintf("❌ Insufficient register values: expected %d, got %d", len(paramColumns), len(values)))
return return
} }
if len(values) > len(paramColumns){ if len(values) > len(paramColumns){
publishError(client, errorTopic, fmt.Sprintf("❌ Too many register values: expected %d, got %d", len(paramColumns), len(values))) logAndPublishError(client, errorTopic, fmt.Sprintf("❌ Too many register values: expected %d, got %d", len(paramColumns), len(values)))
return return
} }
@@ -231,7 +231,7 @@ func messageHandler(client mqtt.Client, msg mqtt.Message) {
_, err = db.Exec(insertQuery, args...) _, err = db.Exec(insertQuery, args...)
if err != nil { if err != nil {
publishError(client, errorTopic, fmt.Sprintf("❌ Failed to insert into mfm_readings: %v", err)) logAndPublishError(client, errorTopic, fmt.Sprintf("❌ Failed to insert into mfm_readings: %v", err))
return return
} }
@@ -241,7 +241,7 @@ func messageHandler(client mqtt.Client, msg mqtt.Message) {
// Publishes error messages to a specific topic // Publishes error messages to a specific topic
func publishError(client mqtt.Client, topic, message string) { func logAndPublishError(client mqtt.Client, topic, message string) {
token := client.Publish(topic, 1, false, message) token := client.Publish(topic, 1, false, message)
token.Wait() token.Wait()
if token.Error() != nil { if token.Error() != nil {
@@ -303,6 +303,10 @@ func main() {
sigChan := make(chan os.Signal, 1) sigChan := make(chan os.Signal, 1)
<-sigChan <-sigChan
} }
func logAndPublishError(client mqtt.Client, errorTopic string, message string) {
log.Printf("[ERROR PUBLISH] Topic: %s | Message: %s", errorTopic, message)
publishError(client, errorTopic, message)
}