poc/mqtt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: poc:860bea9f0ef5a4142ce5fdf0d1ec2feaa55b492c
Branches Tags
poc:master poc:dev-mqtt
...
compare: poc:dev-mqtt
Branches Tags
poc:master poc:dev-mqtt

12 Commits
860bea9f0e ... dev-mqtt

Author SHA1 Message Date
jothi
acec9d72b4 Merge branch 'master' into dev-mqtt 2025-08-06 05:51:44 +00:00
jothi
64a8a32d1a removed unwanted line for checking branch 2025-08-06 05:40:39 +00:00
jothi
89786c9dfa Merge pull request 'Remove commented line for checking git branch' (#1) from dev-mqtt into master 
Reviewed-on: #1
 2025-08-06 05:28:29 +00:00
jothi
bd68af1f58 Remove commented line for checking git branch 2025-08-06 05:17:46 +00:00
jothi
f6d3908088 remove done comment line for checking 2025-07-31 08:19:51 +00:00
jothi
5aad1d7d44 checking git commits 2025-07-31 07:55:32 +00:00
jothi
02fe9ea45e Removed commented l;ines for checking 2025-07-31 04:23:58 +00:00
jothi
edfdc3699d removed commented line 2025-07-31 04:03:16 +00:00
jothi
dbcb05d017 Added comment line for checking 2025-07-30 04:59:36 +00:00
jothi
3fd408d3ee Added commented line for unwanted 2025-07-29 08:43:54 +00:00
jothi
36142de600 Added power house logic latest code 2025-07-29 06:51:57 +00:00
jothi
1c2e45541c Added correct power house logic 2025-07-29 06:20:28 +00:00
1 changed files with 8 additions and 190 deletions
Expand all files Collapse all files

198
mqtt_client.go
View File

@@ -1,174 +1,3 @@
// package main
// // import (
// // "database/sql"
// // "encoding/json"
// // "fmt"
// // "log"
// // "os"
// // mqtt "github.com/eclipse/paho.mqtt.golang"
// // "github.com/joho/godotenv"
// // _ "github.com/lib/pq" // PostgreSQL driver
// // )
// import (
// "database/sql"
// "encoding/json"
// "fmt"
// "log"
// "os"
// mqtt "github.com/eclipse/paho.mqtt.golang"
// "github.com/joho/godotenv"
// _ "github.com/lib/pq"
// )
// var (
// mqttBroker string
// dbDriver string
// dbDataSource string
// clientID string
// )
// func init() {
// // Load environment variables from .env file
// if err := godotenv.Load(); err != nil {
// log.Fatalf("Error loading .env file: %v", err)
// }
// // Retrieve environment variables
// mqttBroker = os.Getenv("MQTT_BROKER")
// dbDriver = os.Getenv("DB_DRIVER")
// dbDataSource = os.Getenv("DB_DATASOURCE")
// clientID = os.Getenv("CLIENT_ID")
// // Validate required environment variables
// if mqttBroker == "" || dbDriver == "" || dbDataSource == "" || clientID == "" {
// log.Fatalf("Missing required environment variables")
// }
// }
// var (
// topics = map[string]byte{"test": 1} // Topics to subscribe with QoS 1
// db *sql.DB // Database connection
// )
// // Handles incoming MQTT messages
// func messageHandler(client mqtt.Client, msg mqtt.Message) {
// fmt.Printf("Received message on topic %s: %s\n", msg.Topic(), msg.Payload())
// requiredFields := []string{"temp", "humidity", "pressure", "timestamp"}
// errorTopic := "error_topic"
// // Validate JSON payload
// var jsonData map[string]interface{}
// if err := json.Unmarshal(msg.Payload(), &jsonData); err != nil {
// publishError(client, errorTopic, fmt.Sprintf("Invalid JSON payload on topic %s: %s", msg.Topic(), msg.Payload()))
// return
// }
// // Check for missing fields
// for _, field := range requiredFields {
// if _, exists := jsonData[field]; !exists {
// publishError(client, errorTopic, fmt.Sprintf("Missing field '%s' in payload on topic %s: %s", field, msg.Topic(), msg.Payload()))
// return
// }
// }
// // Create table dynamically if it doesn't exist
// tableName := msg.Topic()
// createTableQuery := fmt.Sprintf(`
// CREATE TABLE IF NOT EXISTS %s (
// id SERIAL PRIMARY KEY,
// temperature FLOAT NOT NULL,
// humidity INT NOT NULL,
// pressure INT NOT NULL,
// timestamp TIMESTAMP NOT NULL
// )`, tableName)
// if _, err := db.Exec(createTableQuery); err != nil {
// publishError(client, errorTopic, fmt.Sprintf("Failed to create table '%s': %v", tableName, err))
// return
// }
// // Insert data into the table
// insertQuery := fmt.Sprintf(`
// INSERT INTO %s (temperature, humidity, pressure, timestamp)
// VALUES ($1, $2, $3, $4)`, tableName)
// _, err := db.Exec(insertQuery, jsonData["temperature"], jsonData["humidity"], jsonData["pressure"], jsonData["timestamp"])
// if err != nil {
// publishError(client, errorTopic, fmt.Sprintf("Failed to insert data into table '%s': %v", tableName, err))
// } else {
// log.Printf("Data successfully inserted into table '%s'.", tableName)
// }
// }
// // Publishes error messages to a specific topic
// func publishError(client mqtt.Client, topic, message string) {
// token := client.Publish(topic, 1, false, message)
// token.Wait()
// if token.Error() != nil {
// log.Printf("Failed to publish error to topic '%s': %v", topic, token.Error())
// }
// }
// // Handles successful MQTT connection
// func connectHandler(client mqtt.Client) {
// log.Println("Connected to MQTT broker")
// }
// // Handles MQTT connection loss
// func connectionLostHandler(client mqtt.Client, err error) {
// log.Printf("Connection lost: %v. Attempting to reconnect...", err)
// }
// func main() {
// var err error
// godotenv.Load()
// // Connect to the database
// db, err = sql.Open(dbDriver, dbDataSource)
// if err != nil {
// log.Fatalf("Database connection failed: %v", err)
// }
// defer db.Close()
// // Validate database connection
// if err := db.Ping(); err != nil {
// log.Fatalf("Database ping failed: %v", err)
// }
// // Configure MQTT client
// opts := mqtt.NewClientOptions().
// AddBroker(mqttBroker).
// SetClientID(clientID).
// SetUsername("test"). // todo: Change to .env variable
// SetPassword("test"). // todo: Change to .env variable
// SetCleanSession(false).
// SetAutoReconnect(true).
// SetOrderMatters(false).
// SetOnConnectHandler(connectHandler).
// SetConnectionLostHandler(connectionLostHandler)
// client := mqtt.NewClient(opts)
// if token := client.Connect(); token.Wait() && token.Error() != nil {
// log.Fatalf("Failed to connect to MQTT broker: %v", token.Error())
// }
// // Subscribe to topics
// for topic, qos := range topics {
// if token := client.Subscribe(topic, qos, messageHandler); token.Wait() && token.Error() != nil {
// log.Fatalf("Failed to subscribe to topic '%s': %v", topic, token.Error())
// }
// log.Printf("Subscribed to topic: %s", topic)
// }
// // Keep the program running
// sigChan := make(chan os.Signal, 1)
// <-sigChan
// }
//
package main
import (
@@ -214,6 +43,10 @@ var (
topics = map[string]byte{"powerhouse": 1} // Topics to subscribe with QoS 1
db *sql.DB // Database connection
)
func publishError(client mqtt.Client, topic, errorMsg string) {
token := client.Publish(topic, 0, false, errorMsg)
token.Wait()
}
// Handles incoming MQTT messages
func messageHandler(client mqtt.Client, msg mqtt.Message) {
@@ -266,7 +99,7 @@ func messageHandler(client mqtt.Client, msg mqtt.Message) {
return
}
// Check for duplicates
var exists int
existsQuery := `SELECT 1 FROM temp_live_readings WHERE created_at = $1 AND mfm_meter_id = $2 LIMIT 1`
err = db.QueryRow(existsQuery, payload.CreatedAt, mfmMeterID).Scan(&exists)
@@ -294,6 +127,7 @@ func messageHandler(client mqtt.Client, msg mqtt.Message) {
publishError(client, errorTopic, fmt.Sprintf("❌ DB insert failed: %v", err))
return
}
//..
log.Printf("✅ Data inserted successfully for plant ID %d and mfm_meter_id %d", plantID, mfmMeterID)
@@ -371,7 +205,6 @@ func messageHandler(client mqtt.Client, msg mqtt.Message) {
return
}
// Build insert statement
columns := []string{"plant_id", "mfm_meter_id"}
placeholders := []string{"$1", "$2"}
args := []interface{}{plantID, mfmMeterID}
@@ -411,15 +244,6 @@ func messageHandler(client mqtt.Client, msg mqtt.Message) {
}
// Publishes error messages to a specific topic
func publishError(client mqtt.Client, topic, message string) {
token := client.Publish(topic, 1, false, message)
token.Wait()
if token.Error() != nil {
log.Printf("Failed to publish error to topic '%s': %v", topic, token.Error())
}
}
// Handles successful MQTT connection
func connectHandler(client mqtt.Client) {
log.Println("Connected to MQTT broker")
@@ -449,6 +273,8 @@ func main() {
opts := mqtt.NewClientOptions().
AddBroker(mqttBroker).
SetClientID(clientID).
// SetUsername("test").
// SetPassword("test").
SetCleanSession(false).
SetAutoReconnect(true).
SetOrderMatters(false).
@@ -460,7 +286,6 @@ func main() {
log.Fatalf("Failed to connect to MQTT broker: %v", token.Error())
}
// Subscribe to topics
for topic, qos := range topics {
if token := client.Subscribe(topic, qos, messageHandler); token.Wait() && token.Error() != nil {
log.Fatalf("Failed to subscribe to topic '%s': %v", topic, token.Error())
@@ -468,15 +293,8 @@ func main() {
log.Printf("Subscribed to topic: %s", topic)
}
// Keep the program running
sigChan := make(chan os.Signal, 1)
<-sigChan
}