// Publishes a bill over the real broker and waits for the ack, standing in for // a till until one is available to test with. // // go run ./scratch/mqttpub package main import ( "encoding/json" "fmt" "log" "os" "time" mqtt "github.com/eclipse/paho.mqtt.golang" "github.com/joho/godotenv" ) const ( locationID = "1135" terminalID = "T4A9" orderID = "99999999-8888-4777-8666-555555555555" // distinct from the HTTP test ) func main() { _ = godotenv.Load() broker := os.Getenv("MQTT_URL") if broker == "" { broker = "tcp://66.116.225.226:1883" } opts := mqtt.NewClientOptions(). AddBroker(broker). SetClientID("pos-e2e-probe"). SetUsername(os.Getenv("MQTT_USER")). SetPassword(os.Getenv("MQTT_PASSWORD")). SetCleanSession(true) client := mqtt.NewClient(opts) if t := client.Connect(); t.Wait() && t.Error() != nil { log.Fatal("connect:", t.Error()) } defer client.Disconnect(500) fmt.Println("connected to", broker) acks := make(chan []byte, 1) ackTopic := fmt.Sprintf("nearle/pos/%s/%s/ack", locationID, terminalID) if t := client.Subscribe(ackTopic, 1, func(_ mqtt.Client, m mqtt.Message) { acks <- m.Payload() }); t.Wait() && t.Error() != nil { log.Fatal("subscribe:", t.Error()) } fmt.Println("listening on", ackTopic) batch := map[string]any{ "schema": 1, "batch_id": "batch-mqtt-0001", "store_id": locationID, "terminal_id": terminalID, "sent_at": time.Now().UTC().Format(time.RFC3339), "orders": []map[string]any{{ "id": orderID, "invoice_number": "INV-2608-T4A9-00002", "created_at": time.Now().UTC().Format(time.RFC3339), "terminal_id": terminalID, "cashier": "Divya", "customer": map[string]any{"id": "c-2", "mobile": "9840099999", "name": "Ravi"}, "subtotal": 60.0, "discount": 0.0, "promos": []any{}, "tax": 4.44, "tax_breakdown": map[string]float64{"0.08": 4.44}, "round_off": 0.0, "total": 60.0, "points_earned": 0, "points_redeemed": 0, "payments": []map[string]any{{"method": "upi", "amount": 60.0, "reference": "TESTUPI"}}, "items": []map[string]any{{ "product_id": "6988", "barcode": "6988", "name": "Mysore Banana", "quantity": 1, "unit_price": 60.0, "discount": 0.0, "gst_rate": 0.08, "tax": 4.44, "line_total": 60.0, }}, }}, } body, _ := json.Marshal(batch) orderTopic := fmt.Sprintf("nearle/pos/%s/%s/order", locationID, terminalID) if t := client.Publish(orderTopic, 1, false, body); t.Wait() && t.Error() != nil { log.Fatal("publish:", t.Error()) } fmt.Println("published a bill to", orderTopic) // The same 20 seconds a real till waits before giving up and re-sending. select { case payload := <-acks: fmt.Println("\nACK RECEIVED:") var pretty map[string]any _ = json.Unmarshal(payload, &pretty) out, _ := json.MarshalIndent(pretty, " ", " ") fmt.Println(" " + string(out)) case <-time.After(20 * time.Second): fmt.Println("\nNO ACK within 20s — a real till would keep the bill and send it again") os.Exit(1) } // A heartbeat too, so the Redis presence path is exercised. health, _ := json.Marshal(map[string]any{ "schema": 1, "status": "online", "terminal_id": terminalID, "location_id": locationID, "store_name": "Ragul stores Selvapuram", "app_version": "1.1.0", "pending_bills": 0, "pending_registrations": 0, "today_bills": 2, "today_amount": 170.0, "printer_reachable": false, "reported_at": time.Now().UTC().Format(time.RFC3339), }) healthTopic := fmt.Sprintf("nearle/pos/%s/%s/health", locationID, terminalID) if t := client.Publish(healthTopic, 1, false, health); t.Wait() && t.Error() != nil { log.Fatal("publish health:", t.Error()) } fmt.Println("\npublished a heartbeat to", healthTopic) // The registration uplink. Tested over HTTP early on; this is the same // service reached over the broker, which is the path a real till uses. custBatch, _ := json.Marshal(map[string]any{ "schema": 1, "batch_id": "batch-cust-mqtt-0001", "store_id": locationID, "terminal_id": terminalID, "customers": []map[string]any{{ "id": "3d7a0000-0000-4000-8000-000000000001", "mobile": "9840077777", "name": "MQTT Probe Shopper", "registered_at": time.Now().UTC().Format(time.RFC3339), "registered_by_terminal": terminalID, }}, }) custTopic := fmt.Sprintf("nearle/pos/%s/%s/customer", locationID, terminalID) if t := client.Publish(custTopic, 1, false, custBatch); t.Wait() && t.Error() != nil { log.Fatal("publish customer:", t.Error()) } fmt.Println("published a registration to", custTopic) select { case payload := <-acks: fmt.Println(" registration ACK:", string(payload)) case <-time.After(20 * time.Second): fmt.Println(" NO ACK for the registration") os.Exit(1) } }