//go:build zmqtest package main import ( "encoding/json" "flag" "fmt" "log" "math/rand" "sync" "sync/atomic" "time" "github.com/luxfi/czmq/v4" "github.com/luxfi/dex/pkg/lx" ) // TestClient represents a client that sends orders to the multi-node cluster type TestClient struct { dealers []*czmq.Sock nodeAddrs []string ordersent uint64 responsesRcv uint64 mu sync.RWMutex } // NewTestClient creates a new test client func NewTestClient(nodeAddrs []string) (*TestClient, error) { client := &TestClient{ nodeAddrs: nodeAddrs, dealers: make([]*czmq.Sock, len(nodeAddrs)), } // Create DEALER socket for each node for i, addr := range nodeAddrs { dealer := czmq.NewSock(czmq.Dealer) // Set identity for the dealer dealer.SetIdentity(fmt.Sprintf("client-%d-%d", time.Now().Unix(), i)) // Connect to node's ROUTER socket if err := dealer.Connect(addr); err != nil { return nil, fmt.Errorf("failed to connect to %s: %w", addr, err) } client.dealers[i] = dealer } return client, nil } // SendOrder sends an order to a random node func (c *TestClient) SendOrder(order *lx.Order) error { // Select random node for load balancing nodeIdx := rand.Intn(len(c.dealers)) dealer := c.dealers[nodeIdx] // Marshal order to JSON data, err := json.Marshal(order) if err != nil { return fmt.Errorf("failed to marshal order: %w", err) } // Send to dealer (no identity needed, DEALER handles it) dealer.SendFrame([]byte(""), czmq.FlagMore) if err := dealer.SendFrame(data, 0); err != nil { return fmt.Errorf("failed to send order: %w", err) } atomic.AddUint64(&c.ordersent, 1) // Receive response asynchronously go c.receiveResponse(dealer) return nil } // receiveResponse receives response from node func (c *TestClient) receiveResponse(dealer *czmq.Sock) { // Empty delimiter _, _, err := dealer.RecvFrame() if err != nil { return } // Response data data, _, err := dealer.RecvFrame() if err != nil { return } atomic.AddUint64(&c.responsesRcv, 1) var response map[string]interface{} if err := json.Unmarshal(data, &response); err != nil { log.Printf("Failed to unmarshal response: %v", err) return } if trades, ok := response["trades"].([]interface{}); ok && len(trades) > 0 { log.Printf("Order matched! Trades: %v", trades) } } // RunLoadTest runs a load test sending many orders func (c *TestClient) RunLoadTest(duration time.Duration, ordersPerSec int) { log.Printf("Starting load test: %d orders/sec for %v", ordersPerSec, duration) ticker := time.NewTicker(time.Second / time.Duration(ordersPerSec)) defer ticker.Stop() timeout := time.After(duration) for { select { case <-timeout: log.Printf("Load test complete. Sent: %d, Received: %d", atomic.LoadUint64(&c.ordersent), atomic.LoadUint64(&c.responsesRcv)) return case <-ticker.C: // Generate random order order := &lx.Order{ Symbol: "BTC-USD", Side: lx.Side(rand.Intn(2)), Type: lx.Limit, Price: 45000 + float64(rand.Intn(10000)), Size: float64(rand.Intn(10)+1) * 0.1, User: fmt.Sprintf("user-%d", rand.Intn(100)), } if err := c.SendOrder(order); err != nil { log.Printf("Failed to send order: %v", err) } } } } // Close closes all dealer sockets func (c *TestClient) Close() { for _, dealer := range c.dealers { if dealer != nil { dealer.Destroy() } } } // RunTestClient is the main function for the test client func RunTestClient() { var ( nodes = flag.String("nodes", "tcp://localhost:5002,tcp://localhost:5102,tcp://localhost:5202", "Comma-separated node addresses") duration = flag.Duration("duration", 30*time.Second, "Test duration") rate = flag.Int("rate", 100, "Orders per second") ) flag.Parse() // Parse node addresses nodeAddrs := splitAndTrim(*nodes, ",") if len(nodeAddrs) == 0 { log.Fatal("No node addresses provided") } // Create client client, err := NewTestClient(nodeAddrs) if err != nil { log.Fatalf("Failed to create client: %v", err) } defer client.Close() // Run load test client.RunLoadTest(*duration, *rate) // Wait a bit for final responses time.Sleep(2 * time.Second) log.Printf("Final stats - Sent: %d, Received: %d", atomic.LoadUint64(&client.ordersent), atomic.LoadUint64(&client.responsesRcv)) }