Go: XINFO STREAM.

go/api/base_client.go

@@ -6820,6 +6820,128 @@ func (client *baseClient) XRevRangeWithOptions(
 	return handleMapOfArrayOfStringArrayOrNilResponse(result)
 }
 
+// Returns information about the stream stored at `key`.
+//
+// See [valkey.io] for details.
+//
+// Parameters:
+//
+//	key - The key of the stream.
+//
+// Return value:
+//
+//	A stream information for the given `key`. See the example for a sample response.
+//
+// Example:
+//
+//	infoBreef, err := client.XInfoStream(key)
+//	infoBreef:
+//	// map[string]any {
+//	// 	"entries-added" : 1,
+//	// 	"first-entry" : []any{
+//	// 		"1719877599564-0", []any{"some_field", "some_value", ...},
+//	// 	},
+//	// 	"groups" : 1,
+//	// 	"last-entry" : []any{
+//	// 		"1719877599564-1", []any{"some_field", "some_value", ...},
+//	// 	},
+//	// 	"last-generated-id" : "1719877599564-1",
+//	// 	"length" : 1,
+//	// 	"max-deleted-entry-id" : "0-0",
+//	// 	"radix-tree-keys" : 1,
+//	// 	"radix-tree-nodes" : 2,
+//	// 	"recorded-first-entry-id" : "1719877599564-1",
+//	// }
+//
+// [valkey.io]: https://valkey.io/commands/xinfo-stream/
+func (client *baseClient) XInfoStream(key string) (map[string]interface{}, error) {
+	result, err := client.executeCommand(C.XInfoStream, []string{key})
+	if err != nil {
+		return nil, err
+	}
+	return handleStringToAnyMapResponse(result)
+}
+
+// Returns detailed information about the stream stored at `key`.
+//
+// See [valkey.io] for details.
+//
+// Parameters:
+//
+//	key  - The key of the stream.
+//	opts - Stream info options.
+//
+// Return value:
+//
+//	A detailed stream information for the given `key`. See the example for a sample response.
+//
+// Example:
+//
+//	options := options.NewXInfoStreamOptionsOptions().SetCount(5)
+//	infoFull, err := client.XInfoStreamWithOptions(key, options)
+//	infoFull:
+//	// map[string]any {
+//	// 	"entries" : []any{
+//	// 		"1719877599564-0", []any{"some_field", "some_value", ...},
+//	// 		...
+//	//  },
+//	// 	"entries-added" : 2,
+//	// 	"groups" : []any{
+//	// 		map[string]any {
+//	// 			"consumers" : []any{
+//	// 				map[string]any {
+//	// 					"active-time" : 1737592821596,
+//	// 					"name" : "consumer1",
+//	// 					"pel-count" : 1,
+//	// 					"pending" : []any{
+//	// 						[]any{ "1719877599564-0", 1737592821596, 1 },
+//	// 						...
+//	// 					},
+//	// 					"seen-time" : 1737592821596,
+//	// 				},
+//	// 			},
+//	// 			"entries-read" : 1,
+//	// 			"lag" : 1,
+//	// 			"last-delivered-id" : "1719877599564-0",
+//	// 			"name" : "group1"
+//	// 			"pel-count" : 1,
+//	// 			"pending" : []any{
+//	// 				[]any{ "1719877599564-0", "consumer1", 1737592821596, 1 },
+//	// 				...
+//	// 			},
+//	// 		},
+//	// 	},
+//	// 	"last-generated-id" : "1719877599564-1",
+//	// 	"length" : 2,
+//	// 	"max-deleted-entry-id" : "0-0",
+//	// 	"radix-tree-keys" : 1,
+//	// 	"radix-tree-nodes" : 2,
+//	// 	"recorded-first-entry-id" : "1719877599564-1",
+//	// }
+//
+//	// get info for the first consumer of the first group
+//	consumer := infoFull["groups"].([]any)[0].(map[string]any)["consumers"].([]any)[0]
+//
+// [valkey.io]: https://valkey.io/commands/xinfo-stream/
+func (client *baseClient) XInfoStreamFullWithOptions(
+	key string,
+	opts *options.XInfoStreamOptions,
+) (map[string]interface{}, error) {
+	args := []string{key, options.FullKeyword}
+	if opts != nil {
+		optionArgs, err := opts.ToArgs()
+		if err != nil {
+			return nil, err
+		}
+		args = append(args, optionArgs...)
+	}
+	result, err := client.executeCommand(C.XInfoStream, args)
+	if err != nil {
+		return nil, err
+	}
+	return handleStringToAnyMapResponse(result)
+}
+
 // Reads or modifies the array of bits representing the string that is held at key
 // based on the specified sub commands.
 //

go/api/options/constants.go

@@ -4,6 +4,7 @@ package options
 
 const (
 	CountKeyword string = "COUNT"      // Valkey API keyword used to extract specific number of matching indices from a list.
+	FullKeyword  string = "FULL"       // Valkey API keyword used in XINFO STREAM
 	MatchKeyword string = "MATCH"      // Valkey API keyword used to indicate the match filter.
 	NoValue      string = "NOVALUE"    // Valkey API keyword for the no value option for hcsan command.
 	WithScore    string = "WITHSCORE"  // Valkey API keyword for the with score option for zrank and zrevrank commands.

go/api/options/stream_options.go

@@ -399,6 +399,33 @@ func (sco *StreamClaimOptions) ToArgs() ([]string, error) {
 	return optionArgs, nil
 }
 
+// Optional arguments for `XInfoStream` in [StreamCommands]
+type XInfoStreamOptions struct {
+	count int64
+}
+
+// Create new empty `XInfoStreamOptions`
+func NewXInfoStreamOptionsOptions() *XInfoStreamOptions {
+	return &XInfoStreamOptions{-1}
+}
+
+// Request verbose information limiting the returned PEL entries.
+// If `0` is specified, returns verbose information with no limit.
+func (xiso *XInfoStreamOptions) SetCount(count int64) *XInfoStreamOptions {
+	xiso.count = count
+	return xiso
+}
+
+func (xiso *XInfoStreamOptions) ToArgs() ([]string, error) {
+	var args []string
+
+	if xiso.count > -1 {
+		args = append(args, "COUNT", utils.IntToString(xiso.count))
+	}
+
+	return args, nil
+}
+
 type StreamBoundary string
 
 // Create a new stream boundary.
@@ -435,7 +462,7 @@ func (sro *StreamRangeOptions) ToArgs() ([]string, error) {
 	var args []string
 
 	if sro.countIsSet {
-		args = append(args, "COUNT", utils.IntToString(sro.count))
+		args = append(args, CountKeyword, utils.IntToString(sro.count))
 	}
 
 	return args, nil

go/api/response_handlers.go

@@ -1003,3 +1003,18 @@ func handleXPendingDetailResponse(response *C.struct_CommandResponse) ([]XPendin
 
 	return pendingDetails, nil
 }
+
+func handleStringToAnyMapResponse(response *C.struct_CommandResponse) (map[string]interface{}, error) {
+	defer C.free_command_response(response)
+
+	typeErr := checkResponseType(response, C.Map, false)
+	if typeErr != nil {
+		return nil, typeErr
+	}
+
+	result, err := parseMap(response)
+	if err != nil {
+		return nil, err
+	}
+	return result.(map[string]interface{}), nil
+}

go/api/stream_commands.go

@@ -109,6 +109,10 @@ type StreamCommands interface {
 		options *options.StreamClaimOptions,
 	) ([]string, error)
 
+	XInfoStream(key string) (map[string]any, error)
+
+	XInfoStreamFullWithOptions(key string, options *options.XInfoStreamOptions) (map[string]any, error)
+
 	XRange(key string, start options.StreamBoundary, end options.StreamBoundary) (map[string][][]string, error)
 
 	XRangeWithOptions(

go/integTest/shared_commands_test.go

@@ -6712,6 +6712,57 @@ func (suite *GlideTestSuite) TestXGroupStreamCommands() {
 	})
 }
 
+func (suite *GlideTestSuite) TestXInfoStream() {
+	suite.runWithDefaultClients(func(client api.BaseClient) {
+		key := uuid.NewString()
+		group := uuid.NewString()
+		consumer := uuid.NewString()
+
+		xadd, err := client.XAddWithOptions(key, [][]string{{"a", "b"}, {"c", "d"}}, options.NewXAddOptions().SetId("1-0"))
+		assert.Nil(suite.T(), err)
+		assert.Equal(suite.T(), "1-0", xadd.Value())
+
+		suite.verifyOK(client.XGroupCreate(key, group, "0-0"))
+
+		_, err = client.XReadGroup(group, consumer, map[string]string{key: ">"})
+		assert.NoError(suite.T(), err)
+
+		infoSmall, err := client.XInfoStream(key)
+		assert.NoError(suite.T(), err)
+		assert.Equal(suite.T(), int64(1), infoSmall["length"])
+		assert.Equal(suite.T(), int64(1), infoSmall["groups"])
+		expectedEntry := []any{"1-0", []any{"a", "b", "c", "d"}}
+		assert.Equal(suite.T(), expectedEntry, infoSmall["first-entry"])
+		assert.Equal(suite.T(), expectedEntry, infoSmall["last-entry"])
+
+		xadd, err = client.XAddWithOptions(key, [][]string{{"e", "f"}}, options.NewXAddOptions().SetId("1-1"))
+		assert.Nil(suite.T(), err)
+		assert.Equal(suite.T(), "1-1", xadd.Value())
+
+		infoFull, err := client.XInfoStreamFullWithOptions(key, options.NewXInfoStreamOptionsOptions().SetCount(1))
+		assert.NoError(suite.T(), err)
+		assert.Equal(suite.T(), int64(2), infoFull["length"])
+
+		if suite.serverVersion >= "7.0.0" {
+			assert.Equal(suite.T(), "1-0", infoFull["recorded-first-entry-id"])
+		} else {
+			assert.NotContains(suite.T(), infoFull, "recorded-first-entry-id")
+			assert.NotContains(suite.T(), infoFull, "max-deleted-entry-id")
+			assert.NotContains(suite.T(), infoFull, "entries-added")
+			assert.NotContains(suite.T(), infoFull["groups"].([]any)[0], "entries-read")
+			assert.NotContains(suite.T(), infoFull["groups"].([]any)[0], "lag")
+		}
+		// first consumer of first group
+		cns := infoFull["groups"].([]any)[0].(map[string]any)["consumers"].([]any)[0]
+		assert.Contains(suite.T(), cns, "seen-time")
+		if suite.serverVersion >= "7.2.0" {
+			assert.Contains(suite.T(), cns, "active-time")
+		} else {
+			assert.NotContains(suite.T(), cns, "active-time")
+		}
+	})
+}
+
 func (suite *GlideTestSuite) TestSetBit_SetSingleBit() {
 	suite.runWithDefaultClients(func(client api.BaseClient) {
 		key := uuid.New().String()