194 lines
7.4 KiB
Go
194 lines
7.4 KiB
Go
// Package simulator executes only harmless ping/version commands in a bounded worker.
|
|
package simulator
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/json"
|
|
"errors"
|
|
"time"
|
|
|
|
"github.com/eclipse/paho.golang/paho"
|
|
"robot-market/test-mqtt-5/internal/backend"
|
|
"robot-market/test-mqtt-5/internal/contracts"
|
|
"robot-market/test-mqtt-5/internal/events"
|
|
"robot-market/test-mqtt-5/internal/mqtt"
|
|
)
|
|
|
|
type entry struct {
|
|
fingerprint [32]byte
|
|
command contracts.Command
|
|
correlation []byte
|
|
ack contracts.Ack
|
|
due, forget time.Time
|
|
complete bool
|
|
}
|
|
type Simulator struct {
|
|
Client *mqtt.Client
|
|
BootID string
|
|
cache map[string]*entry
|
|
Executions uint64
|
|
sequence uint64
|
|
}
|
|
|
|
func New(client *mqtt.Client) *Simulator {
|
|
return &Simulator{Client: client, BootID: backend.ID(), cache: make(map[string]*entry)}
|
|
}
|
|
func (s *Simulator) envelope() contracts.Envelope {
|
|
return contracts.Envelope{SchemaVersion: 1, DeviceID: s.Client.Config.Topics.DeviceID}
|
|
}
|
|
func (s *Simulator) status(state string, will bool) contracts.Status {
|
|
status := contracts.Status{Envelope: s.envelope(), BootID: s.BootID, State: state, Implementation: "go-simulator", ReportedProtocol: "5"}
|
|
if will {
|
|
status.WillConfiguredAt = backend.Timestamp(time.Now())
|
|
} else {
|
|
status.Timestamp = backend.Timestamp(time.Now())
|
|
}
|
|
return status
|
|
}
|
|
func (s *Simulator) Will() *paho.WillMessage {
|
|
p := mqtt.Packet(mqtt.Topic(s.Client.Config, "status"), 1, true, s.status("offline", true))
|
|
return &paho.WillMessage{Topic: p.Topic, Payload: p.Payload, QoS: 1, Retain: true}
|
|
}
|
|
func (s *Simulator) ack(id, stage, code string) contracts.Ack {
|
|
ok := stage == "accepted" || stage == "completed"
|
|
ack := contracts.Ack{Envelope: s.envelope(), RequestID: id, Timestamp: backend.Timestamp(time.Now()), Stage: stage, OK: &ok}
|
|
if !ok {
|
|
ack.Error = &contracts.AckError{Code: code, Message: "test simulator rejected the request"}
|
|
}
|
|
return ack
|
|
}
|
|
func (s *Simulator) sendAck(ctx context.Context, ack contracts.Ack, correlation []byte) {
|
|
p := mqtt.Packet(mqtt.Topic(s.Client.Config, "acks"), 1, false, ack)
|
|
p.Properties.CorrelationData = append([]byte(nil), correlation...)
|
|
if err := s.Client.Publish(ctx, p); err != nil {
|
|
s.Client.Record(events.Kind("ack_not_confirmed"), events.Internal, p, ack.RequestID, "cached outcome; replay only on duplicate request")
|
|
}
|
|
}
|
|
func (s *Simulator) purge(now time.Time) {
|
|
for id, e := range s.cache {
|
|
if e.complete && !now.Before(e.forget) {
|
|
delete(s.cache, id)
|
|
}
|
|
}
|
|
}
|
|
func (s *Simulator) Handle(ctx context.Context, p *paho.Publish) {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
value, err := contracts.Decode("commands", p.Payload, s.Client.Config.Topics.DeviceID)
|
|
if err != nil || p.Topic != mqtt.Topic(s.Client.Config, "commands") {
|
|
s.Client.Record(events.ValidationFailed, events.In, p, "", "invalid command contract")
|
|
return
|
|
}
|
|
cmd := *value.(*contracts.Command)
|
|
s.Client.Record(events.MessageReceived, events.In, p, cmd.RequestID, "")
|
|
if p.Retain || p.QoS != 1 || p.Properties == nil || p.Properties.ResponseTopic != mqtt.Topic(s.Client.Config, "acks") || (len(p.Properties.CorrelationData) == 0 || len(p.Properties.CorrelationData) > 64) || p.Properties.MessageExpiry == nil || *p.Properties.MessageExpiry == 0 || *p.Properties.MessageExpiry > 60 {
|
|
s.Client.Record(events.ValidationFailed, events.In, p, cmd.RequestID, "requires non-retained QoS1 and exact native command properties")
|
|
return
|
|
}
|
|
now := time.Now()
|
|
s.purge(now)
|
|
data, _ := json.Marshal(cmd)
|
|
fingerprint := sha256.Sum256(append(data, p.Properties.CorrelationData...))
|
|
if previous, ok := s.cache[cmd.RequestID]; ok {
|
|
if previous.fingerprint != fingerprint {
|
|
s.sendAck(ctx, s.ack(cmd.RequestID, "rejected", "conflicting_request"), p.Properties.CorrelationData)
|
|
return
|
|
}
|
|
s.Client.Record(events.DuplicateRequest, events.In, p, cmd.RequestID, "cached ACK replay; no repeated execution")
|
|
s.sendAck(ctx, previous.ack, previous.correlation)
|
|
return
|
|
}
|
|
if err := cmd.CheckDeadline(now); err != nil {
|
|
s.sendAck(ctx, s.ack(cmd.RequestID, "rejected", "deadline_invalid"), p.Properties.CorrelationData)
|
|
return
|
|
}
|
|
if len(s.cache) >= s.Client.Config.Limits.DedupEntries {
|
|
s.sendAck(ctx, s.ack(cmd.RequestID, "rejected", "dedup_capacity"), p.Properties.CorrelationData)
|
|
return
|
|
}
|
|
accepted := s.ack(cmd.RequestID, "accepted", "")
|
|
s.cache[cmd.RequestID] = &entry{fingerprint: fingerprint, command: cmd, correlation: append([]byte(nil), p.Properties.CorrelationData...), ack: accepted, due: now.Add(s.Client.Config.Simulator.AckDelay), forget: now.Add(s.Client.Config.Limits.DedupRetention)}
|
|
s.Client.Record(events.CommandAccepted, events.Internal, p, cmd.RequestID, "admitted to bounded simulator worker")
|
|
s.sendAck(ctx, accepted, p.Properties.CorrelationData)
|
|
}
|
|
func (s *Simulator) CompleteDue(ctx context.Context, now time.Time) {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
s.purge(now)
|
|
for id, e := range s.cache {
|
|
if e.complete || now.Before(e.due) {
|
|
continue
|
|
}
|
|
if err := e.command.CheckDeadline(now); err != nil {
|
|
e.ack = s.ack(id, "failed", "deadline_elapsed")
|
|
} else {
|
|
s.Executions++
|
|
e.ack = s.ack(id, "completed", "")
|
|
if e.command.Command == "ping" {
|
|
yes := true
|
|
e.ack.Result = &contracts.Result{Pong: &yes}
|
|
} else {
|
|
e.ack.Result = &contracts.Result{Version: "test-sim-1.0.0"}
|
|
}
|
|
}
|
|
e.complete = true
|
|
e.forget = now.Add(s.Client.Config.Limits.DedupRetention)
|
|
s.Client.Events.Add(events.Event{Kind: events.Kind("simulator_execution"), Direction: events.Internal, DeviceID: s.Client.Config.Topics.DeviceID, RequestID: id, Outcome: e.ack.Stage})
|
|
s.sendAck(ctx, e.ack, e.correlation)
|
|
}
|
|
}
|
|
func (s *Simulator) Run(ctx context.Context) (runErr error) {
|
|
if err := s.Client.Start(ctx, []string{"commands"}, s.Will()); err != nil {
|
|
return err
|
|
}
|
|
defer func() {
|
|
shutdown, stop := context.WithTimeout(context.Background(), s.Client.Config.Deadlines.Shutdown)
|
|
defer stop()
|
|
if err := s.Client.Publish(shutdown, mqtt.Packet(mqtt.Topic(s.Client.Config, "status"), 1, true, s.status("offline", false))); err != nil {
|
|
s.Client.Record(events.Kind("offline_not_confirmed"), events.Internal, nil, "", "graceful offline status not confirmed")
|
|
}
|
|
if err := s.Client.CloseContext(shutdown); err != nil && runErr == nil {
|
|
runErr = err
|
|
}
|
|
}()
|
|
tick := time.NewTicker(50 * time.Millisecond)
|
|
defer tick.Stop()
|
|
telemetry := time.NewTicker(s.Client.Config.Simulator.TelemetryInterval)
|
|
defer telemetry.Stop()
|
|
online := false
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case err := <-s.Client.Fatal():
|
|
return err
|
|
case <-s.Client.Down:
|
|
online = false
|
|
case p := <-s.Client.Incoming:
|
|
s.Handle(ctx, p)
|
|
case now := <-tick.C:
|
|
if !s.Client.Ready() {
|
|
online = false
|
|
}
|
|
if !online && s.Client.Ready() {
|
|
if err := s.Client.Publish(ctx, mqtt.Packet(mqtt.Topic(s.Client.Config, "status"), 1, true, s.status("online", false))); err != nil {
|
|
return errors.New("online status not confirmed")
|
|
}
|
|
online = true
|
|
}
|
|
s.CompleteDue(ctx, now)
|
|
case <-telemetry.C:
|
|
if !s.Client.Ready() {
|
|
continue
|
|
}
|
|
s.sequence++
|
|
seq := s.sequence
|
|
message := contracts.Telemetry{Envelope: s.envelope(), MessageID: backend.ID(), BootID: s.BootID, Sequence: &seq, Timestamp: backend.Timestamp(time.Now()), Readings: map[string]float64{"temperature_c": 22.5, "executions": float64(s.Executions)}}
|
|
_ = s.Client.Publish(ctx, mqtt.Packet(mqtt.Topic(s.Client.Config, "telemetry"), byte(s.Client.Config.MQTT.QoS), false, message))
|
|
}
|
|
}
|
|
}
|