WebSocket 测试

复制import asyncio
import websockets

async def test_websocket():
    uri = "wss://echo.websocket.org"
    async with websockets.connect(uri) as ws:
        # 发送消息
        await ws.send("Hello WebSocket")
        # 接收回显
        response = await ws.recv()
        print(f"Received: {response}")

        # 心跳测试：发送 ping 并等待 pong
        pong = await ws.ping()
        await pong
        print("Pong received, connection alive")

asyncio.run(test_websocket())

复制package main

import (
	"fmt"
	"net/url"
	"github.com/gorilla/websocket"
	"log"
)

func main() {
	u := url.URL{Scheme: "wss", Host: "echo.websocket.org", Path: ""}
	c, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
	if err != nil {
		log.Fatal("dial:", err)
	}
	defer c.Close()

	// 发送消息
	err = c.WriteMessage(websocket.TextMessage, []byte("Hello WebSocket"))
	if err != nil {
		log.Fatal("write:", err)
	}

	// 接收回显
	_, message, err := c.ReadMessage()
	if err != nil {
		log.Fatal("read:", err)
	}
	fmt.Printf("Received: %s\n", message)

	// 心跳测试：发送 Pong 帧（服务端通常回复 Ping）
	err = c.WriteMessage(websocket.PongMessage, []byte("keepalive"))
	if err != nil {
		log.Fatal("pong:", err)
	}
	fmt.Println("Heartbeat sent")
}

复制const ws = new WebSocket('wss://echo.websocket.org');

ws.onopen = () => {
    console.log('Connected');
    // 发送消息
    ws.send('Hello WebSocket');

    // 心跳：发送 ping（浏览器 WebSocket API 自动处理 pong）
    // 可手动发送空消息作为心跳
    setInterval(() => {
        if (ws.readyState === WebSocket.OPEN) {
            ws.send('ping');
        }
    }, 30000);
};

ws.onmessage = (event) => {
    console.log('Received:', event.data);
};

ws.onerror = (error) => {
    console.error('WebSocket error:', error);
};