Respond to signal ping / pong (#871)

* Respond to signal ping / pong

* Pass back 1 for pong for now, we don't need the timestamp

* update protocol

test/client/client.go

@@ -55,6 +55,7 @@ type RTCClient struct {
 	lossyDCSub          *webrtc.DataChannel
 	publisherConnected  atomic.Bool
 	publisherNegotiated atomic.Bool
+	pongReceivedAt      atomic.Int64
 
 	// tracks waiting to be acked, cid => trackInfo
 	pendingPublishedTracks map[string]*livekit.TrackInfo
@@ -347,6 +348,8 @@ func (c *RTCClient) Run() error {
 			delete(c.trackSenders, sid)
 			delete(c.localTracks, sid)
 			c.lock.Unlock()
+		case *livekit.SignalResponse_Pong:
+			c.pongReceivedAt.Store(msg.Pong)
 		}
 	}
 }
@@ -441,6 +444,18 @@ func (c *RTCClient) RefreshToken() string {
 	return c.refreshToken
 }
 
+func (c *RTCClient) PongReceivedAt() int64 {
+	return c.pongReceivedAt.Load()
+}
+
+func (c *RTCClient) SendPing() error {
+	return c.SendRequest(&livekit.SignalRequest{
+		Message: &livekit.SignalRequest_Ping{
+			Ping: time.Now().UnixNano(),
+		},
+	})
+}
+
 func (c *RTCClient) SendRequest(msg *livekit.SignalRequest) error {
 	payload, err := proto.Marshal(msg)
 	if err != nil {

test/singlenode_test.go

@@ -328,6 +328,23 @@ func TestSingleNodeCORS(t *testing.T) {
 	require.Equal(t, "testhost.com", res.Header.Get("Access-Control-Allow-Origin"))
 }
 
+func TestPingPong(t *testing.T) {
+	if testing.Short() {
+		t.SkipNow()
+		return
+	}
+	_, finish := setupSingleNodeTest("TestPingPong")
+	defer finish()
+
+	c1 := createRTCClient("c1", defaultServerPort, nil)
+	waitUntilConnected(t, c1)
+
+	require.NoError(t, c1.SendPing())
+	require.Eventually(t, func() bool {
+		return c1.PongReceivedAt() > 0
+	}, time.Second, 10*time.Millisecond)
+}
+
 func TestSingleNodeJoinAfterClose(t *testing.T) {
 	if testing.Short() {
 		t.SkipNow()