Go: XADD

go/api/base_client.go

@@ -11,6 +11,7 @@ import "C"
 
 import (
 	"errors"
+	"fmt"
 	"math"
 	"strconv"
 	"unsafe"
@@ -26,6 +27,7 @@ type BaseClient interface {
 	HashCommands
 	ListCommands
 	SetCommands
+	StreamCommands
 	ConnectionManagementCommands
 	GenericBaseCommands
 	// Close terminates the client by closing all associated resources.
@@ -1204,3 +1206,23 @@ func (client *baseClient) Renamenx(key string, newKey string) (Result[bool], err
 	}
 	return handleBooleanResponse(result)
 }
+
+func (client *baseClient) XAdd(key string, values [][]string) (Result[string], error) {
+	args := make([]string, 0, 2+2*len(values))
+	args = append(args, key, "*")
+	for _, pair := range values {
+		if len(pair) != 2 {
+			return CreateNilStringResult(), fmt.Errorf(
+				"Array entry had the wrong length. Expected length 2 but got length %d",
+				len(pair),
+			)
+		}
+		args = append(args, pair...)
+	}
+
+	result, err := client.executeCommand(C.XAdd, args)
+	if err != nil {
+		return CreateNilStringResult(), err
+	}
+	return handleStringResponse(result)
+}

go/api/stream_commands.go

@@ -0,0 +1,29 @@
+// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
+
+package api
+
+// Supports commands and transactions for the "Stream Commands" group for standalone and cluster clients.
+//
+// See [valkey.io] for details.
+//
+// [valkey.io]: https://valkey.io/commands/#stream
+type StreamCommands interface {
+	// Adds an entry to the specified stream stored at `key`. If the `key` doesn't exist, the stream is created.
+	//
+	// See [valkey.io] for details.
+	//
+	// Parameters:
+	//  key      - The key of the stream.
+	//  values   - Field-value pairs to be added to the entry.
+	//
+	// Return value:
+	//  The id of the added entry.
+	//
+	// For example:
+	//  result, err := client.XAdd("myStream", [][]string{{"field1", "value1"}, {"field2", "value2"}})
+	//  result.IsNil(): false
+	//  result.Value(): "1526919030474-55"
+	//
+	// [valkey.io]: https://valkey.io/commands/xadd/
+	XAdd(key string, values [][]string) (Result[string], error)
+}

go/integTest/shared_commands_test.go

@@ -3743,7 +3743,7 @@ func (suite *GlideTestSuite) TestUnlink() {
 	})
 }
 
-func (suite *GlideTestSuite) Test_Rename() {
+func (suite *GlideTestSuite) TestRename() {
 	suite.runWithDefaultClients(func(client api.BaseClient) {
 		// Test 1 Check if the command successfully renamed
 		key := "{keyName}" + uuid.NewString()
@@ -3781,3 +3781,31 @@ func (suite *GlideTestSuite) TestRenamenx() {
 		assert.Equal(suite.T(), false, res2.Value())
 	})
 }
+
+func (suite *GlideTestSuite) TestXAdd() {
+	suite.runWithDefaultClients(func(client api.BaseClient) {
+		key := uuid.NewString()
+		// stream does not exist
+		res, err := client.XAdd(key, [][]string{{"field1", "value1"}, {"field1", "value2"}})
+		assert.Nil(suite.T(), err)
+		assert.False(suite.T(), res.IsNil())
+		// don't check the value, because it contains server's timestamp
+
+		// adding data to existing stream
+		res, err = client.XAdd(key, [][]string{{"field3", "value3"}})
+		assert.Nil(suite.T(), err)
+		assert.False(suite.T(), res.IsNil())
+
+		// incorrect input
+		_, err = client.XAdd(key, [][]string{})
+		assert.NotNil(suite.T(), err)
+		_, err = client.XAdd(key, [][]string{{"1", "2", "3"}})
+		assert.NotNil(suite.T(), err)
+
+		// key is not a string
+		key = uuid.NewString()
+		client.Set(key, "abc")
+		_, err = client.XAdd(key, [][]string{{"f", "v"}})
+		assert.NotNil(suite.T(), err)
+	})
+}