From 8e84523d4bfd7a07f083e35910fe81ef91d46868 Mon Sep 17 00:00:00 2001 From: Overlord Date: Mon, 8 Dec 2025 06:58:43 +0100 Subject: [PATCH] continuous send for simulators --- bot.go | 120 ++++++++++++++++++++++++++++++++++++++-------------- sim.go | 131 +++++++++++++++++++++++++++++++++++++++++++-------------- 2 files changed, 188 insertions(+), 63 deletions(-) diff --git a/bot.go b/bot.go index b4069ae..8dd5130 100644 --- a/bot.go +++ b/bot.go @@ -5,10 +5,14 @@ package main import ( "encoding/json" "flag" + "fmt" "log" + "math/rand" "net/url" "os" "os/signal" + "sync" + "sync/atomic" "syscall" "time" @@ -18,8 +22,12 @@ import ( const ( gatewayURL = "ws://localhost:3333/sync" apiKey = "gateway" - // must match the mod's channel id used by your mod simulator - channelID = "123456789" + channelID = "123456789" +) + +var ( + minInterval = 500 * time.Millisecond + maxInterval = 5 * time.Second ) type Handshake struct { @@ -28,7 +36,7 @@ type Handshake struct { } type BotHandshake struct { - ChannelId string `json:"channel_id"` // match gateway field exactly + ChannelId string `json:"channel_id"` } type GatewayAck struct { @@ -56,12 +64,21 @@ type GatewayMessageIn struct { ReceivedAt time.Time `json:"-"` } +func randomDuration(min, max time.Duration) time.Duration { + if max <= min { + return min + } + diff := int64(max - min) + n := rand.Int63n(diff) + return min + time.Duration(n) +} + func main() { - var ( - botID = flag.String("bot", "sim-bot-1", "bot id") - sendAfter = flag.Duration("send-after", 0, "optional: send a bot->mod message after this delay (e.g. 2s)") - sendMsg = flag.String("msg", "Hello from bot!", "optional bot->mod test message content") - ) + rand.Seed(time.Now().UnixNano()) + + botID := flag.String("bot", "sim-bot-1", "bot id") + sendAfter := flag.Duration("send-after", 0, "optional: send a bot->mod message after this delay (e.g. 2s)") + sendMsg := flag.String("msg", "Hello from bot!", "optional bot->mod test message content") flag.Parse() interrupt := make(chan os.Signal, 1) @@ -75,21 +92,25 @@ func main() { q.Set("api_key", apiKey) u.RawQuery = q.Encode() - log.Printf("Connecting to %s", u.String()) conn, _, err := websocket.DefaultDialer.Dial(u.String(), nil) if err != nil { log.Fatalf("Failed to connect: %v", err) } - // we intentionally don't defer conn.Close() here immediately; we'll do on shutdown - log.Println("Connected to gateway") + defer conn.Close() + + var writeMu sync.Mutex - // handle server pings by replying a Pong (safe) conn.SetPingHandler(func(appData string) error { - log.Println("Received ping from server, sending pong") - return conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(5*time.Second)) + writeMu.Lock() + err := conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(5*time.Second)) + writeMu.Unlock() + if err != nil { + log.Printf("Failed to send pong: %v", err) + return err + } + return nil }) - // send bot handshake (must at least include bot_id) bhs := BotHandshake{ChannelId: channelID} data, err := json.Marshal(bhs) if err != nil { @@ -97,21 +118,22 @@ func main() { log.Fatalf("Failed to marshal bot handshake: %v", err) } hs := Handshake{Type: "bot", Data: data} + + writeMu.Lock() if err := conn.WriteJSON(hs); err != nil { + writeMu.Unlock() _ = conn.Close() log.Fatalf("Failed to send handshake: %v", err) } - log.Println("Handshake sent (bot)") + writeMu.Unlock() - // Instead of ReadJSON, read raw message so we can see whatever the server returns (or why it closes). - // This avoids a silent parsing failure if server returns non-JSON or closes immediately. conn.SetReadDeadline(time.Now().Add(10 * time.Second)) msgType, raw, err := conn.ReadMessage() if err != nil { _ = conn.Close() log.Fatalf("Failed to read handshake response: %v", err) } - _ = conn.SetReadDeadline(time.Time{}) // clear deadline + _ = conn.SetReadDeadline(time.Time{}) if msgType == websocket.TextMessage || msgType == websocket.BinaryMessage { log.Printf("Raw handshake reply: %s", string(raw)) @@ -122,11 +144,11 @@ func main() { log.Printf("Parsed ack: status=%q, type=%q", ack.Status, ack.Type) } } else { - log.Printf("Handshake reply was control frame or unexpected type=%d", msgType) + log.Printf("Handshake reply type=%d", msgType) } - // From here, start the normal read loop. done := make(chan struct{}) + go func() { defer close(done) for { @@ -139,20 +161,20 @@ func main() { } return } - if msgType == websocket.TextMessage || msgType == websocket.BinaryMessage { log.Printf("Received from gateway: %s", string(message)) } } }() - // optionally send a bot->mod message after a delay + var msgCounter uint64 = 1 + if *sendAfter > 0 { go func() { time.Sleep(*sendAfter) msg := GatewayMessageIn{ - MsgID: "bot-msg-001", - ID: channelID, // bot reports channel id as ID + MsgID: fmt.Sprintf("bot-msg-%06d", atomic.AddUint64(&msgCounter, 1)), + ID: channelID, Author: User{ ID: *botID, Name: "SimBot", @@ -160,20 +182,55 @@ func main() { Content: *sendMsg, Ts: time.Now().UTC(), } - // set a write deadline + writeMu.Lock() _ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second)) if err := conn.WriteJSON(msg); err != nil { log.Printf("Failed to send bot->mod test message: %v", err) - return + } else { + log.Printf("Sent bot->mod test message (channel=%s)", channelID) } - log.Printf("Sent bot->mod test message (channel=%s)", channelID) _ = conn.SetWriteDeadline(time.Time{}) + writeMu.Unlock() }() } - log.Println("Bot simulator running. Press Ctrl+C to exit.") + go func() { + for { + select { + case <-done: + return + default: + } + d := randomDuration(minInterval, maxInterval) + select { + case <-done: + return + case <-time.After(d): + msgNum := atomic.AddUint64(&msgCounter, 1) + msg := GatewayMessageIn{ + MsgID: fmt.Sprintf("sim-bot-msg-%06d", msgNum), + ID: channelID, + Author: User{ + ID: fmt.Sprintf("%s-%d", *botID, msgNum%1000), + Name: fmt.Sprintf("SimBot%d", msgNum%1000), + }, + Content: fmt.Sprintf("Automated bot message #%d (delay %s)", msgNum, d), + Ts: time.Now().UTC(), + } + writeMu.Lock() + _ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second)) + if err := conn.WriteJSON(msg); err != nil { + writeMu.Unlock() + log.Printf("Failed to send automated bot message: %v", err) + return + } + _ = conn.SetWriteDeadline(time.Time{}) + writeMu.Unlock() + log.Printf("Sent automated bot message %s", msg.MsgID) + } + } + }() - // Wait for interrupt or read loop done for { select { case <-done: @@ -182,8 +239,9 @@ func main() { return case <-interrupt: log.Println("Interrupt received, closing connection...") - // politely close + writeMu.Lock() _ = conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")) + writeMu.Unlock() select { case <-done: case <-time.After(time.Second): diff --git a/sim.go b/sim.go index ce2a105..ccc4aeb 100644 --- a/sim.go +++ b/sim.go @@ -4,10 +4,14 @@ package main import ( "encoding/json" + "fmt" "log" + "math/rand" "net/url" "os" "os/signal" + "sync" + "sync/atomic" "syscall" "time" @@ -22,6 +26,12 @@ const ( channelID = "123456789" ) +// send interval range (random between minInterval and maxInterval) +var ( + minInterval = 500 * time.Millisecond + maxInterval = 5 * time.Second +) + type Handshake struct { Type string `json:"type"` Data json.RawMessage `json:"data"` @@ -59,6 +69,8 @@ type GatewayMessageIn struct { } func main() { + rand.Seed(time.Now().UnixNano()) + interrupt := make(chan os.Signal, 1) signal.Notify(interrupt, os.Interrupt, syscall.SIGTERM) @@ -80,19 +92,19 @@ func main() { log.Println("Connected to gateway") - // respond to pings (server ping -> client must pong). Using SetPingHandler is fine, - // but WriteControl for Pong is acceptable too. + var writeMu sync.Mutex + conn.SetPingHandler(func(appData string) error { log.Println("Received ping from server, sending pong") - err := conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(5*time.Second)) - if err != nil { + writeMu.Lock() + defer writeMu.Unlock() + if err := conn.WriteControl(websocket.PongMessage, []byte(appData), time.Now().Add(5*time.Second)); err != nil { log.Printf("Failed to send pong: %v", err) return err } return nil }) - // Build and send handshake including channel id modHS := ModHandshake{ ServerID: serverID, ChannelID: channelID, @@ -107,12 +119,14 @@ func main() { Data: modHSData, } + writeMu.Lock() if err := conn.WriteJSON(handshake); err != nil { + writeMu.Unlock() log.Fatalf("Failed to send handshake: %v", err) } + writeMu.Unlock() log.Println("Handshake sent") - // Read acknowledgment (some servers might not reply with JSON; handle errors) var ack GatewayAck if err := conn.ReadJSON(&ack); err != nil { log.Fatalf("Failed to read acknowledgment: %v", err) @@ -129,62 +143,115 @@ func main() { if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) { log.Printf("WebSocket error: %v", err) } else { - log.Printf("Connection closed: %v", err) + log.Printf("Connection closed/read error: %v", err) } return } - if messageType == websocket.TextMessage || messageType == websocket.BinaryMessage { + switch messageType { + case websocket.TextMessage, websocket.BinaryMessage: log.Printf("Received from server: %s", string(message)) + default: } } }() - // Optional: send a test message after connecting - time.Sleep(1 * time.Second) - testMsg := GatewayMessageIn{ - MsgID: "test-msg-001", - ID: serverID, - Destination: Destination{ - ID: channelID, - }, - Author: User{ - ID: "player-uuid-123", - Name: "TestPlayer", - }, - Content: "Hello from simulated mod!", - Ts: time.Now().UTC(), - } + var msgCounter uint64 = 1 - if err := conn.WriteJSON(testMsg); err != nil { - log.Printf("Failed to send test message: %v", err) - } else { - log.Println("Sent test message to gateway") - } + func() { + testMsg := GatewayMessageIn{ + MsgID: fmt.Sprintf("test-msg-%06d", atomic.AddUint64(&msgCounter, 1)), + ID: serverID, + Destination: Destination{ + ID: channelID, + }, + Author: User{ + ID: "player-uuid-123", + Name: "TestPlayer", + }, + Content: "Hello from simulated mod!", + Ts: time.Now().UTC(), + } - log.Println("Connection established. Responding to pings. Press Ctrl+C to disconnect.") + writeMu.Lock() + if err := conn.WriteJSON(testMsg); err != nil { + log.Printf("Failed to send test message: %v", err) + } else { + log.Println("Sent initial test message to gateway") + } + writeMu.Unlock() + }() + + go func() { + for { + d := randomDuration(minInterval, maxInterval) + + select { + case <-done: + return + case <-time.After(d): + // build message + msgNum := atomic.AddUint64(&msgCounter, 1) + msg := GatewayMessageIn{ + MsgID: fmt.Sprintf("sim-msg-%06d", msgNum), + ID: serverID, + Destination: Destination{ + ID: channelID, + }, + Author: User{ + ID: fmt.Sprintf("sim-user-%d", msgNum%1000), + Name: fmt.Sprintf("SimUser%d", msgNum%1000), + }, + Content: fmt.Sprintf("Random interval message #%d (delay %s)", msgNum, d), + Ts: time.Now().UTC(), + } + + writeMu.Lock() + if err := conn.WriteJSON(msg); err != nil { + log.Printf("Failed to send simulated message: %v", err) + writeMu.Unlock() + return + } + writeMu.Unlock() + + log.Printf("Sent simulated message %s (next wait up to %s)", msg.MsgID, maxInterval) + } + } + }() + + log.Println("Connection established. Sending simulated messages at random intervals. Press Ctrl+C to disconnect.") for { select { case <-done: - log.Println("Connection closed") + log.Println("Connection read loop closed, exiting") return case <-interrupt: log.Println("Interrupt received, closing connection...") + writeMu.Lock() err := conn.WriteMessage( websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""), ) + writeMu.Unlock() if err != nil { log.Printf("Write close error: %v", err) - return } select { case <-done: - case <-time.After(time.Second): + case <-time.After(1 * time.Second): } return } } } + +func randomDuration(min, max time.Duration) time.Duration { + if max <= min { + return min + } + diff := int64(max - min) + n := rand.Int63n(diff) + return min + time.Duration(n) +}