Change to using sobek instead of goja

We are moving to a fork of goja under grafana org called sobek.

More info in:
- grafana/k6#3772
- grafana/k6#3773

go.mod

@@ -1,10 +1,10 @@
 module github.com/mostafa/xk6-kafka
 
-go 1.22
+go 1.20
 
 require (
 	github.com/aws/aws-sdk-go-v2/config v1.18.39
-	github.com/dop251/goja v0.0.0-20240220182346-e401ed450204
+	github.com/grafana/sobek v0.0.0-20240607083612-4f0cd64f4e78
 	github.com/linkedin/goavro/v2 v2.12.0
 	github.com/pavlo-v-chernykh/keystore-go/v4 v4.5.0
 	github.com/riferrei/srclient v0.6.0
@@ -13,12 +13,11 @@ require (
 	github.com/segmentio/kafka-go/sasl/aws_msk_iam_v2 v0.1.0
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.9.0
-	go.k6.io/k6 v0.50.0
+	go.k6.io/k6 v0.51.1-0.20240610082146-1f01a9bc2365
 	gopkg.in/guregu/null.v3 v3.5.0
 )
 
 require (
-	github.com/andybalholm/brotli v1.1.0 // indirect
 	github.com/aws/aws-sdk-go-v2 v1.24.0 // indirect
 	github.com/aws/aws-sdk-go-v2/credentials v1.13.37 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.13.11 // indirect
@@ -33,11 +32,12 @@ require (
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dlclark/regexp2 v1.11.0 // indirect
+	github.com/dop251/goja v0.0.0-20240516125602-ccbae20bcec2 // indirect
+	github.com/evanw/esbuild v0.21.2 // indirect
 	github.com/fatih/color v1.16.0 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-sourcemap/sourcemap v2.1.4+incompatible // indirect
-	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/pprof v0.0.0-20240402174815-29b9bb013b0f // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1 // indirect
@@ -47,6 +47,7 @@ require (
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mstoykov/atlas v0.0.0-20220811071828-388f114305dd // indirect
+	github.com/mstoykov/k6-taskqueue-lib v0.1.0 // indirect
 	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/onsi/gomega v1.20.2 // indirect
 	github.com/pierrec/lz4/v4 v4.1.19 // indirect
@@ -64,15 +65,15 @@ require (
 	go.opentelemetry.io/otel/sdk v1.24.0 // indirect
 	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.1.0 // indirect
-	golang.org/x/crypto v0.21.0 // indirect
-	golang.org/x/net v0.23.0 // indirect
-	golang.org/x/sync v0.6.0 // indirect
-	golang.org/x/sys v0.18.0 // indirect
-	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/crypto v0.24.0 // indirect
+	golang.org/x/net v0.26.0 // indirect
+	golang.org/x/sync v0.7.0 // indirect
+	golang.org/x/sys v0.21.0 // indirect
+	golang.org/x/text v0.16.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240401170217-c3f982113cda // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240401170217-c3f982113cda // indirect
-	google.golang.org/grpc v1.62.1 // indirect
+	google.golang.org/grpc v1.63.2 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

jks.go

@@ -7,7 +7,7 @@ import (
 	"log"
 	"os"
 
-	"github.com/dop251/goja"
+	"github.com/grafana/sobek"
 	"github.com/pavlo-v-chernykh/keystore-go/v4"
 	"go.k6.io/k6/js/common"
 )
@@ -117,7 +117,7 @@ func (*Kafka) loadJKS(jksConfig *JKSConfig) (*JKS, *Xk6KafkaError) {
 	}, nil
 }
 
-func (k *Kafka) loadJKSFunction(call goja.FunctionCall) goja.Value {
+func (k *Kafka) loadJKSFunction(call sobek.FunctionCall) sobek.Value {
 	runtime := k.vu.Runtime()
 	var jksConfig *JKSConfig
 
@@ -171,7 +171,7 @@ func saveServerCaFile(filename string, cert *keystore.TrustedCertificateEntry) *
 		Bytes: cert.Certificate.Content,
 	})
 
-	if err := os.WriteFile(filename, certPem, 0644); err != nil {
+	if err := os.WriteFile(filename, certPem, 0o644); err != nil {
 		return NewXk6KafkaError(
 			failedWriteServerCaFile, "Failed to write CA file", err)
 	}
@@ -186,7 +186,7 @@ func saveClientKeyFile(filename string, key *keystore.PrivateKeyEntry) *Xk6Kafka
 		Bytes: key.PrivateKey,
 	})
 
-	if err := os.WriteFile(filename, keyPem, 0644); err != nil {
+	if err := os.WriteFile(filename, keyPem, 0o644); err != nil {
 		return NewXk6KafkaError(
 			failedWriteKeyFile, "Failed to write key file", err)
 	}
@@ -200,7 +200,7 @@ func saveClientCertFile(filename string, cert *keystore.Certificate) *Xk6KafkaEr
 		Type:  "CERTIFICATE",
 		Bytes: cert.Content,
 	})
-	if err := os.WriteFile(filename, clientCertPem, 0644); err != nil {
+	if err := os.WriteFile(filename, clientCertPem, 0o644); err != nil {
 		return NewXk6KafkaError(
 			failedWriteCertFile, "Failed to write cert file", err)
 	}

jks_test.go

@@ -4,7 +4,7 @@ import (
 	"os"
 	"testing"
 
-	"github.com/dop251/goja"
+	"github.com/grafana/sobek"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -118,8 +118,8 @@ func TestJKS(t *testing.T) {
 func TestLoadJKS_Function(t *testing.T) {
 	test := getTestModuleInstance(t)
 
-	jks := test.module.Kafka.loadJKSFunction(goja.FunctionCall{
-		Arguments: []goja.Value{
+	jks := test.module.Kafka.loadJKSFunction(sobek.FunctionCall{
+		Arguments: []sobek.Value{
 			test.module.vu.Runtime().ToValue(
 				map[string]interface{}{
 					"path":              "fixtures/kafka-keystore.jks",

kafka_helpers_test.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"testing"
 
-	"github.com/dop251/goja"
+	"github.com/grafana/sobek"
 	kafkago "github.com/segmentio/kafka-go"
 	"github.com/stretchr/testify/require"
 	"go.k6.io/k6/js/common"
@@ -20,7 +20,7 @@ const (
 
 // struct to keep all the things test need in one place.
 type kafkaTest struct {
-	rt            *goja.Runtime
+	rt            *sobek.Runtime
 	module        *Module
 	vu            *modulestest.VU
 	samples       chan metrics.SampleContainer
@@ -31,7 +31,7 @@ type kafkaTest struct {
 // nolint: golint,revive
 func getTestModuleInstance(tb testing.TB) *kafkaTest {
 	tb.Helper()
-	runtime := goja.New()
+	runtime := sobek.New()
 	runtime.SetFieldNameMapper(common.FieldNameMapper{})
 
 	ctx, cancel := context.WithCancel(context.Background())

module.go

@@ -4,7 +4,7 @@ import (
 	"crypto/tls"
 	"time"
 
-	"github.com/dop251/goja"
+	"github.com/grafana/sobek"
 	"github.com/riferrei/srclient"
 	kafkago "github.com/segmentio/kafka-go"
 	"github.com/segmentio/kafka-go/compress"
@@ -74,7 +74,7 @@ type (
 	Kafka struct {
 		vu          modules.VU
 		metrics     kafkaMetrics
-		exports     *goja.Object
+		exports     *sobek.Object
 		schemaCache map[string]*Schema
 	}
 	RootModule struct{}
@@ -151,7 +151,7 @@ func (m *Module) defineConstants() {
 	runtime := m.vu.Runtime()
 	mustAddProp := func(name string, val interface{}) {
 		err := m.exports.DefineDataProperty(
-			name, runtime.ToValue(val), goja.FLAG_FALSE, goja.FLAG_FALSE, goja.FLAG_TRUE,
+			name, runtime.ToValue(val), sobek.FLAG_FALSE, sobek.FLAG_FALSE, sobek.FLAG_TRUE,
 		)
 		if err != nil {
 			common.Throw(runtime, err)

reader.go

@@ -7,7 +7,7 @@ import (
 	"io"
 	"time"
 
-	"github.com/dop251/goja"
+	"github.com/grafana/sobek"
 	kafkago "github.com/segmentio/kafka-go"
 	"go.k6.io/k6/js/common"
 	"go.k6.io/k6/metrics"
@@ -109,7 +109,7 @@ func (d *Duration) UnmarshalJSON(b []byte) error {
 // readerClass is a wrapper around kafkago.reader and acts as a JS constructor
 // for this extension, thus it must be called with new operator, e.g. new Reader(...).
 // nolint: funlen
-func (k *Kafka) readerClass(call goja.ConstructorCall) *goja.Object {
+func (k *Kafka) readerClass(call sobek.ConstructorCall) *sobek.Object {
 	runtime := k.vu.Runtime()
 	var readerConfig *ReaderConfig
 	if len(call.Arguments) == 0 {
@@ -134,7 +134,7 @@ func (k *Kafka) readerClass(call goja.ConstructorCall) *goja.Object {
 		common.Throw(runtime, err)
 	}
 
-	err := readerObject.Set("consume", func(call goja.FunctionCall) goja.Value {
+	err := readerObject.Set("consume", func(call sobek.FunctionCall) sobek.Value {
 		var consumeConfig *ConsumeConfig
 		if len(call.Arguments) == 0 {
 			common.Throw(runtime, ErrNotEnoughArguments)
@@ -157,12 +157,12 @@ func (k *Kafka) readerClass(call goja.ConstructorCall) *goja.Object {
 	}
 
 	// This is unnecessary, but it's here for reference purposes
-	err = readerObject.Set("close", func(call goja.FunctionCall) goja.Value {
+	err = readerObject.Set("close", func(call sobek.FunctionCall) sobek.Value {
 		if err := reader.Close(); err != nil {
 			common.Throw(runtime, err)
 		}
 
-		return goja.Undefined()
+		return sobek.Undefined()
 	})
 	if err != nil {
 		common.Throw(runtime, err)

reader_test.go

@@ -5,7 +5,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/dop251/goja"
+	"github.com/grafana/sobek"
 	"github.com/riferrei/srclient"
 	kafkago "github.com/segmentio/kafka-go"
 	"github.com/stretchr/testify/assert"
@@ -329,8 +329,8 @@ func TestReaderClass(t *testing.T) {
 	})
 
 	assert.NotPanics(t, func() {
-		reader := test.module.readerClass(goja.ConstructorCall{
-			Arguments: []goja.Value{
+		reader := test.module.readerClass(sobek.ConstructorCall{
+			Arguments: []sobek.Value{
 				test.module.vu.Runtime().ToValue(
 					map[string]interface{}{
 						"brokers": []string{"localhost:9092"},
@@ -347,9 +347,9 @@ func TestReaderClass(t *testing.T) {
 		assert.Equal(t, this.Config().Topic, "test-reader-class")
 		assert.Equal(t, this.Config().MaxWait, time.Second*3)
 
-		consume := reader.Get("consume").Export().(func(goja.FunctionCall) goja.Value)
-		messages := consume(goja.FunctionCall{
-			Arguments: []goja.Value{
+		consume := reader.Get("consume").Export().(func(sobek.FunctionCall) sobek.Value)
+		messages := consume(sobek.FunctionCall{
+			Arguments: []sobek.Value{
 				test.module.vu.Runtime().ToValue(
 					map[string]interface{}{
 						"limit": 1,
@@ -370,9 +370,9 @@ func TestReaderClass(t *testing.T) {
 		assert.Equal(t, "value", deserializedValue)
 
 		// Close the reader.
-		close := reader.Get("close").Export().(func(goja.FunctionCall) goja.Value)
+		close := reader.Get("close").Export().(func(sobek.FunctionCall) sobek.Value)
 		assert.NotNil(t, close)
-		result := close(goja.FunctionCall{}).Export()
+		result := close(sobek.FunctionCall{}).Export()
 		assert.Nil(t, result)
 
 		// Check if one message was consumed.

schema_registry.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"net/http"
 
-	"github.com/dop251/goja"
+	"github.com/grafana/sobek"
 	"github.com/linkedin/goavro/v2"
 	"github.com/riferrei/srclient"
 	"github.com/santhosh-tekuri/jsonschema/v5"
@@ -92,7 +92,7 @@ func (s *Schema) JsonSchema() *jsonschema.Schema {
 	return s.jsonSchema
 }
 
-func (k *Kafka) schemaRegistryClientClass(call goja.ConstructorCall) *goja.Object {
+func (k *Kafka) schemaRegistryClientClass(call sobek.ConstructorCall) *sobek.Object {
 	runtime := k.vu.Runtime()
 	var configuration *SchemaRegistryConfig
 	var schemaRegistryClient *srclient.SchemaRegistryClient
@@ -117,7 +117,7 @@ func (k *Kafka) schemaRegistryClientClass(call goja.ConstructorCall) *goja.Objec
 		common.Throw(runtime, err)
 	}
 
-	err := schemaRegistryClientObject.Set("getSchema", func(call goja.FunctionCall) goja.Value {
+	err := schemaRegistryClientObject.Set("getSchema", func(call sobek.FunctionCall) sobek.Value {
 		if len(call.Arguments) == 0 {
 			common.Throw(runtime, ErrNotEnoughArguments)
 		}
@@ -143,7 +143,7 @@ func (k *Kafka) schemaRegistryClientClass(call goja.ConstructorCall) *goja.Objec
 		common.Throw(runtime, err)
 	}
 
-	err = schemaRegistryClientObject.Set("createSchema", func(call goja.FunctionCall) goja.Value {
+	err = schemaRegistryClientObject.Set("createSchema", func(call sobek.FunctionCall) sobek.Value {
 		if len(call.Arguments) == 0 {
 			common.Throw(runtime, ErrNotEnoughArguments)
 		}
@@ -170,7 +170,7 @@ func (k *Kafka) schemaRegistryClientClass(call goja.ConstructorCall) *goja.Objec
 	}
 
 	var subjectNameConfig *SubjectNameConfig
-	err = schemaRegistryClientObject.Set("getSubjectName", func(call goja.FunctionCall) goja.Value {
+	err = schemaRegistryClientObject.Set("getSubjectName", func(call sobek.FunctionCall) sobek.Value {
 		if len(call.Arguments) == 0 {
 			common.Throw(runtime, ErrNotEnoughArguments)
 		}
@@ -191,7 +191,7 @@ func (k *Kafka) schemaRegistryClientClass(call goja.ConstructorCall) *goja.Objec
 		common.Throw(runtime, err)
 	}
 
-	err = schemaRegistryClientObject.Set("serialize", func(call goja.FunctionCall) goja.Value {
+	err = schemaRegistryClientObject.Set("serialize", func(call sobek.FunctionCall) sobek.Value {
 		if len(call.Arguments) == 0 {
 			common.Throw(runtime, ErrNotEnoughArguments)
 		}
@@ -213,7 +213,7 @@ func (k *Kafka) schemaRegistryClientClass(call goja.ConstructorCall) *goja.Objec
 		common.Throw(runtime, err)
 	}
 
-	err = schemaRegistryClientObject.Set("deserialize", func(call goja.FunctionCall) goja.Value {
+	err = schemaRegistryClientObject.Set("deserialize", func(call sobek.FunctionCall) sobek.Value {
 		if len(call.Arguments) == 0 {
 			common.Throw(runtime, ErrNotEnoughArguments)
 		}