You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 

103 lines
2.7 KiB

// Command mqtt_washingmachine serves laundry status from sampled MQTT power data.
package main
import (
"context"
"errors"
"fmt"
"log/slog"
"net"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"mqtt_washingmachine/internal/sensor"
"mqtt_washingmachine/internal/web"
)
type config struct {
sensor.MQTTConfig
HTTPAddr string
SampleInterval time.Duration
}
// Empty variables use defaults. SAMPLE_INTERVAL accepts positive Go durations;
// it controls backend sampling only, while the page always polls every 60 s.
func loadConfig() (config, error) {
env := func(key, fallback string) string {
if value := os.Getenv(key); value != "" {
return value
}
return fallback
}
interval, err := time.ParseDuration(env("SAMPLE_INTERVAL", "60s"))
if err != nil || interval <= 0 {
return config{}, errors.New("SAMPLE_INTERVAL must be a positive Go duration, such as 60s")
}
return config{
MQTTConfig: sensor.MQTTConfig{
Broker: env("MQTT_BROKER", "tcp://192.168.1.11:1883"),
Topic: env("MQTT_TOPIC", "tele/tasmota_washingmashine_C21183/SENSOR"),
Username: os.Getenv("MQTT_USERNAME"),
Password: os.Getenv("MQTT_PASSWORD"),
},
HTTPAddr: env("HTTP_ADDR", ":8080"),
SampleInterval: interval,
}, nil
}
func run(ctx context.Context, cfg config) error {
listener, err := net.Listen("tcp", cfg.HTTPAddr)
if err != nil {
return fmt.Errorf("listen for HTTP: %w", err)
}
defer listener.Close()
ctx, cancel := context.WithCancel(ctx)
ticker := time.NewTicker(cfg.SampleInterval)
defer ticker.Stop()
var sampler sensor.Sampler
done := make(chan struct{})
go func() {
defer close(done)
sampler.Run(ctx, ticker.C)
}()
defer func() { cancel(); <-done }()
client := sensor.StartMQTT(cfg.MQTTConfig, &sampler, slog.Default())
defer client.Disconnect(250)
server := &http.Server{
Handler: web.NewHandler(sampler.Status),
ReadHeaderTimeout: 5 * time.Second,
WriteTimeout: 10 * time.Second,
IdleTimeout: 60 * time.Second,
}
errorsCh := make(chan error, 1)
go func() { errorsCh <- server.Serve(listener) }()
slog.Info("Laundry status server listening", "address", listener.Addr().String())
select {
case err := <-errorsCh:
return err
case <-ctx.Done():
shutdownCtx, stop := context.WithTimeout(context.Background(), 5*time.Second)
defer stop()
if err := server.Shutdown(shutdownCtx); err != nil {
server.Close()
return fmt.Errorf("HTTP shutdown: %w", err)
}
return nil
}
}
func main() {
cfg, err := loadConfig()
if err == nil {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
err = run(ctx, cfg)
}
if err != nil {
slog.Error("Laundry status service stopped", "error", err)
os.Exit(1)
}
}