Integrate breakpoint client

browser/breakpoint.go

@@ -7,14 +7,35 @@ import (
 	"log"
 	"net/url"
 	"slices"
+	"strings"
 	"sync"
 	"time"
 
 	"github.com/gorilla/websocket"
-
-	k6modules "go.k6.io/k6/js/modules"
+	"github.com/grafana/xk6-browser/env"
 )
 
+/*
+Protocol:
+
+- get_breakpoints: Client initially requests the breakpoints from the server.
+	- Example: [{"file":"file:///Users/inanc/grafana/k6browser/main/examples/fillform.js", "line": 28}]
+- update_breakpoints: Server sends the updated breakpoints to the client.
+	- Example: [{"file":"file:///Users/inanc/grafana/k6browser/main/examples/fillform.js", "line": 32}]
+- resume: Server sends a message to the client to resume the script execution.
+	- Example: {"command":"resume"}
+
+Client:
+
+- The client pauses the script execution when a breakpoint is hit.
+- The server should send the "resume" message to the client to resume the script execution.
+- The client continuously listens for messages from the server.
+
+Example Run:
+
+- K6_BROWSER_BREAKPOINT_SERVER_URL=ws://localhost:8080/breakpoint k6 run script.js
+*/
+
 type breakpoint struct {
 	File      string `json:"file"`
 	Line      int    `json:"line"`
@@ -27,18 +48,18 @@ type breakpointRegistry struct {
 	pauser        chan chan struct{}
 }
 
-func newBreakpointRegistry(_ k6modules.VU) *breakpointRegistry {
+func newBreakpointRegistry() *breakpointRegistry {
 	return &breakpointRegistry{
-		breakpoints: []breakpoint{
-			{
-				File: "file:///Users/inanc/grafana/k6browser/main/examples/fillform.js",
-				Line: 26,
-			},
-			{
-				File: "file:///Users/inanc/grafana/k6browser/main/examples/fillform.js",
-				Line: 32,
-			},
-		},
+		// breakpoints: []breakpoint{
+		// 	{
+		// 		File: "file:///Users/inanc/grafana/k6browser/main/examples/fillform.js",
+		// 		Line: 26,
+		// 	},
+		// 	{
+		// 		File: "file:///Users/inanc/grafana/k6browser/main/examples/fillform.js",
+		// 		Line: 32,
+		// 	},
+		// },
 		pauser: make(chan chan struct{}, 1),
 	}
 }
@@ -76,6 +97,9 @@ func (br *breakpointRegistry) resume() {
 // when a breakpoint is hit in the script.
 func pauseOnBreakpoint(vu moduleVU) {
 	bp := vu.breakpointRegistry
+	if bp == nil { // breakpoints are disabled
+		return
+	}
 
 	pos := getCurrentLineNumber(vu)
 	log.Printf("current line: %v", pos)
@@ -151,7 +175,7 @@ func (bc *breakpointClient) listen() {
 
 		switch envelope.Command {
 		case "update_breakpoints":
-			bc.handleUpdateBreakpoints(envelope.Data)
+			bc.updateBreakpoints(envelope.Data)
 		case "resume":
 			bc.handleResume()
 		default:
@@ -160,7 +184,7 @@ func (bc *breakpointClient) listen() {
 	}
 }
 
-func (bc *breakpointClient) handleUpdateBreakpoints(data []byte) {
+func (bc *breakpointClient) updateBreakpoints(data []byte) {
 	var breakpoints []breakpoint
 	if err := json.Unmarshal(data, &breakpoints); err != nil {
 		log.Printf("breakpointClient: parsing breakpoints: %v", err)
@@ -173,9 +197,35 @@ func (bc *breakpointClient) handleResume() {
 	bc.registry.resume()
 }
 
+// updateInitialBreakpoints requests the initial breakpoints from the server.
+// It blocks until the server responds.
+// TODO: Add context to the function.
+func (bc *breakpointClient) updateInitialBreakpoints() error {
+	envelope := map[string]any{
+		"command": "get_breakpoints",
+	}
+	message, err := json.Marshal(envelope)
+	if err != nil {
+		return fmt.Errorf("breakpointClient: marshaling get_breakpoints message: %w", err)
+	}
+	if err := bc.conn.WriteMessage(websocket.TextMessage, message); err != nil {
+		return fmt.Errorf("breakpointClient: sending get_breakpoints message: %w", err)
+	}
+
+	// wait for the server to respond
+	if _, message, err = bc.conn.ReadMessage(); err != nil {
+		return fmt.Errorf("breakpointClient: reading get_breakpoints response: %w", err)
+	}
+	log.Println("breakpointClient: received initial breakpoints:", string(message))
+
+	bc.updateBreakpoints(message)
+
+	return nil
+}
+
 func (bc *breakpointClient) sendPause() error {
 	envelope := map[string]any{
-		"type": "pause",
+		"command": "pause",
 	}
 	message, err := json.Marshal(envelope)
 	if err != nil {
@@ -200,3 +250,8 @@ func (bc *breakpointClient) close() error {
 	}
 	return nil
 }
+
+func parseBreakpointServerURL(envLookup env.LookupFunc) string {
+	v, _ := envLookup(env.BreakpointServerURL)
+	return strings.TrimSpace(v)
+}

browser/breakpoint_test.go

@@ -105,7 +105,7 @@ func TestBreakpointClient_SendPause(t *testing.T) {
 		err = json.Unmarshal(message, &envelope)
 		require.NoError(t, err)
 
-		assert.Equal(t, "pause", envelope["type"])
+		assert.Equal(t, "pause", envelope["command"])
 	})
 
 	require.NoError(t, client.sendPause())

browser/module.go

@@ -8,6 +8,7 @@ package browser
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"log"
 	"net/http"
@@ -33,13 +34,14 @@ type (
 	// RootModule is the global module instance that will create module
 	// instances for each VU.
 	RootModule struct {
-		PidRegistry    *pidRegistry
-		remoteRegistry *remoteRegistry
-		initOnce       *sync.Once
-		tracesMetadata map[string]string
-		filePersister  filePersister
-		testRunID      string
-		isSync         bool // remove later
+		PidRegistry        *pidRegistry
+		remoteRegistry     *remoteRegistry
+		initOnce           *sync.Once
+		tracesMetadata     map[string]string
+		filePersister      filePersister
+		breakpointRegistry *breakpointRegistry
+		testRunID          string
+		isSync             bool // remove later
 	}
 
 	// JSModule exposes the properties available to the JS script.
@@ -63,8 +65,9 @@ var (
 // New returns a pointer to a new RootModule instance.
 func New() *RootModule {
 	return &RootModule{
-		PidRegistry: &pidRegistry{},
-		initOnce:    &sync.Once{},
+		PidRegistry:        &pidRegistry{},
+		breakpointRegistry: newBreakpointRegistry(),
+		initOnce:           &sync.Once{},
 	}
 }
 
@@ -73,9 +76,10 @@ func New() *RootModule {
 // JS API.
 func NewSync() *RootModule {
 	return &RootModule{
-		PidRegistry: &pidRegistry{},
-		initOnce:    &sync.Once{},
-		isSync:      true,
+		PidRegistry:        &pidRegistry{},
+		breakpointRegistry: newBreakpointRegistry(),
+		initOnce:           &sync.Once{},
+		isSync:             true,
 	}
 }
 
@@ -111,7 +115,7 @@ func (m *RootModule) NewModuleInstance(vu k6modules.VU) k6modules.Instance {
 					m.tracesMetadata,
 				),
 				taskQueueRegistry:  newTaskQueueRegistry(vu),
-				breakpointRegistry: newBreakpointRegistry(vu),
+				breakpointRegistry: m.breakpointRegistry,
 				filePersister:      m.filePersister,
 				testRunID:          m.testRunID,
 			}),
@@ -142,6 +146,19 @@ func (m *RootModule) initialize(vu k6modules.VU) {
 	if err != nil {
 		k6ext.Abort(vu.Context(), "parsing browser traces metadata: %v", err)
 	}
+	if uri := parseBreakpointServerURL(initEnv.LookupEnv); uri != "" {
+		client, err := dialBreakpointServer(vu.Context(), uri, m.breakpointRegistry)
+		if err != nil {
+			// TODO: Use k6ext.Abort instead of panic. But it somehow fails
+			// with a nil pointer dereference.
+			panic(fmt.Errorf("dialing breakpoint server: %w", err))
+		}
+		if err := client.updateInitialBreakpoints(); err != nil {
+			panic(fmt.Errorf("updating initial breakpoints: %w", err))
+		}
+		go client.listen()
+	}
+
 	if _, ok := initEnv.LookupEnv(env.EnableProfiling); ok {
 		go startDebugServer()
 	}

env/env.go

@@ -63,6 +63,10 @@ const (
 	// filter the browser logs based on their category. It supports
 	// regular expressions.
 	LogCategoryFilter = "K6_BROWSER_LOG_CATEGORY_FILTER"
+
+	// BreakpointServerURL is an environment variable that can be used to
+	// set the URL of the breakpoint server.
+	BreakpointServerURL = "K6_BROWSER_BREAKPOINT_SERVER_URL"
 )
 
 // Tracing.