Go: XINFO CONSUMERS. (valkey-io#3120)

* Go: `XINFO CONSUMERS`.

Signed-off-by: Yury-Fridlyand <yury.fridlyand@improving.com>

Author: Yury-Fridlyand

CHANGELOG.md

@@ -4,6 +4,7 @@
 * Go: Add ZRangeStore ([3105](https://github.com/valkey-io/valkey-glide/pull/3105))
 * Go: Add `ZUNION` ([#3119](https://github.com/valkey-io/valkey-glide/pull/3119))
 * Go: Add `ZUNIONSTORE` ([#3136](https://github.com/valkey-io/valkey-glide/pull/3136))
+* Go: Add `XINFO CONSUMERS` ([#3120](https://github.com/valkey-io/valkey-glide/pull/3120))
 * Go: Add `XINFO GROUPS` ([#3106](https://github.com/valkey-io/valkey-glide/pull/3106))
 * Go: Add `ZInterCard` ([#3078](https://github.com/valkey-io/valkey-glide/issues/3078))
 

go/api/base_client.go

@@ -5813,6 +5813,30 @@ func (client *baseClient) XInfoStreamFullWithOptions(
 	return handleStringToAnyMapResponse(result)
 }
 
+// Returns the list of all consumers and their attributes for the given consumer group of the
+// stream stored at `key`.
+//
+// See [valkey.io] for details.
+//
+// Parameters:
+//
+//	key   - The key of the stream.
+//	group - The consumer group name.
+//
+// Return value:
+//
+//	An array of [api.XInfoConsumerInfo], where each element contains the attributes
+//	of a consumer for the given consumer group of the stream at `key`.
+//
+// [valkey.io]: https://valkey.io/commands/xinfo-consumers/
+func (client *baseClient) XInfoConsumers(key string, group string) ([]XInfoConsumerInfo, error) {
+	response, err := client.executeCommand(C.XInfoConsumers, []string{key, group})
+	if err != nil {
+		return nil, err
+	}
+	return handleXInfoConsumersResponse(response)
+}
+
 // Returns the list of all consumer groups and their attributes for the stream stored at `key`.
 //
 // See [valkey.io] for details.

go/api/response_handlers.go

@@ -1178,6 +1178,49 @@ func handleXPendingDetailResponse(response *C.struct_CommandResponse) ([]XPendin
 	return pendingDetails, nil
 }
 
+func handleXInfoConsumersResponse(response *C.struct_CommandResponse) ([]XInfoConsumerInfo, error) {
+	defer C.free_command_response(response)
+
+	typeErr := checkResponseType(response, C.Array, false)
+	if typeErr != nil {
+		return nil, typeErr
+	}
+	arrData, err := parseArray(response)
+	if err != nil {
+		return nil, err
+	}
+	converted, err := arrayConverter[map[string]interface{}]{
+		nil,
+		false,
+	}.convert(arrData)
+	if err != nil {
+		return nil, err
+	}
+	arr, ok := converted.([]map[string]interface{})
+	if !ok {
+		return nil, &errors.RequestError{Msg: fmt.Sprintf("unexpected type: %T", converted)}
+	}
+
+	result := make([]XInfoConsumerInfo, 0, len(arr))
+
+	for _, group := range arr {
+		info := XInfoConsumerInfo{
+			Name:    group["name"].(string),
+			Pending: group["pending"].(int64),
+			Idle:    group["idle"].(int64),
+		}
+		switch inactive := group["inactive"].(type) {
+		case int64:
+			info.Inactive = CreateInt64Result(inactive)
+		default:
+			info.Inactive = CreateNilInt64Result()
+		}
+		result = append(result, info)
+	}
+
+	return result, nil
+}
+
 func handleXInfoGroupsResponse(response *C.struct_CommandResponse) ([]XInfoGroupInfo, error) {
 	defer C.free_command_response(response)
 

go/api/response_types.go

@@ -244,6 +244,21 @@ func CreateNilXPendingSummary() XPendingSummary {
 	return XPendingSummary{0, CreateNilStringResult(), CreateNilStringResult(), make([]ConsumerPendingMessage, 0)}
 }
 
+// XInfoConsumerInfo represents a group information returned by `XInfoConsumers` command.
+type XInfoConsumerInfo struct {
+	// The consumer's name.
+	Name string
+	// The number of entries in the PEL: pending messages for the consumer, which are messages that were delivered but are yet
+	// to be acknowledged.
+	Pending int64
+	// The number of milliseconds that have passed since the consumer's last attempted interaction (Examples: XREADGROUP,
+	// XCLAIM, XAUTOCLAIM).
+	Idle int64
+	// The number of milliseconds that have passed since the consumer's last successful interaction (Examples: XREADGROUP that
+	// actually read some entries into the PEL, XCLAIM/XAUTOCLAIM that actually claimed some entries).
+	Inactive Result[int64]
+}
+
 // XInfoGroupInfo represents a group information returned by `XInfoGroups` command.
 type XInfoGroupInfo struct {
 	// The consumer group's name.

go/api/stream_commands.go

@@ -113,6 +113,8 @@ type StreamCommands interface {
 
 	XInfoStreamFullWithOptions(key string, options *options.XInfoStreamOptions) (map[string]any, error)
 
+	XInfoConsumers(key string, group string) ([]XInfoConsumerInfo, error)
+
 	XInfoGroups(key string) ([]XInfoGroupInfo, error)
 
 	XRange(key string, start options.StreamBoundary, end options.StreamBoundary) ([]XRangeResponse, error)

go/api/stream_commands_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"regexp"
 	"strings"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/valkey-io/valkey-glide/go/api/options"
@@ -1778,6 +1779,71 @@ func ExampleGlideClusterClient_XInfoStreamFullWithOptions() {
 	// }
 }
 
+func ExampleGlideClient_XInfoConsumers() {
+	var client *GlideClient = getExampleGlideClient() // example helper function
+	key := uuid.NewString()
+	group := "myGroup"
+
+	// create an empty stream with a group
+	client.XGroupCreateWithOptions(key, group, "0-0", *options.NewXGroupCreateOptions().SetMakeStream())
+	// add couple of entries
+	client.XAddWithOptions(key, [][]string{{"e1_f1", "e1_v1"}, {"e1_f2", "e1_v2"}}, *options.NewXAddOptions().SetId("0-1"))
+	client.XAddWithOptions(key, [][]string{{"e2_f1", "e2_v1"}, {"e2_f2", "e2_v2"}}, *options.NewXAddOptions().SetId("0-2"))
+	// read them
+	client.XReadGroup(group, "myConsumer", map[string]string{key: ">"})
+	// get the info
+	time.Sleep(100 * time.Millisecond)
+	response, err := client.XInfoConsumers(key, group)
+	if err != nil {
+		fmt.Println("Glide example failed with an error: ", err)
+	}
+
+	// Expanded:
+	fmt.Printf("Consumer name:  %s\n", response[0].Name)
+	fmt.Printf("PEL count:      %d\n", response[0].Pending)
+	// exact values of `Idle` and `Inactive` depend on timing
+	fmt.Printf("Idle > 0:       %t\n", response[0].Idle > 0)
+	fmt.Printf("Inactive > 0:   %t\n", response[0].Inactive.Value() > 0) // Added in version 7.0.0
+	// Output:
+	// Consumer name:  myConsumer
+	// PEL count:      2
+	// Idle > 0:       true
+	// Inactive > 0:   true
+}
+
+func ExampleGlideClusterClient_XInfoConsumers() {
+	var client *GlideClusterClient = getExampleGlideClusterClient() // example helper function
+	key := uuid.NewString()
+	group := "myGroup"
+	consumer := "myConsumer"
+
+	// create an empty stream with a group
+	client.XGroupCreateWithOptions(key, group, "0-0", *options.NewXGroupCreateOptions().SetMakeStream())
+	// add couple of entries
+	client.XAddWithOptions(key, [][]string{{"e1_f1", "e1_v1"}, {"e1_f2", "e1_v2"}}, *options.NewXAddOptions().SetId("0-1"))
+	client.XAddWithOptions(key, [][]string{{"e2_f1", "e2_v1"}, {"e2_f2", "e2_v2"}}, *options.NewXAddOptions().SetId("0-2"))
+	// read them
+	client.XReadGroupWithOptions(group, consumer, map[string]string{key: ">"}, *options.NewXReadGroupOptions().SetCount(1))
+	// get the info
+	time.Sleep(100 * time.Millisecond)
+	response, err := client.XInfoConsumers(key, group)
+	if err != nil {
+		fmt.Println("Glide example failed with an error: ", err)
+	}
+
+	// Expanded:
+	fmt.Printf("Consumer name:  %s\n", response[0].Name)
+	fmt.Printf("PEL count:      %d\n", response[0].Pending)
+	// exact values of `Idle` and `Inactive` depend on timing
+	fmt.Printf("Idle > 0:       %t\n", response[0].Idle > 0)
+	fmt.Printf("Inactive > 0:   %t\n", response[0].Inactive.Value() > 0) // Added in version 7.0.0
+	// Output:
+	// Consumer name:  myConsumer
+	// PEL count:      1
+	// Idle > 0:       true
+	// Inactive > 0:   true
+}
+
 func ExampleGlideClient_XInfoGroups() {
 	var client *GlideClient = getExampleGlideClient() // example helper function
 	key := uuid.NewString()

go/integTest/shared_commands_test.go

@@ -7124,6 +7124,106 @@ func (suite *GlideTestSuite) TestXInfoStream() {
 	})
 }
 
+func (suite *GlideTestSuite) TestXInfoConsumers() {
+	suite.runWithDefaultClients(func(client api.BaseClient) {
+		key := uuid.NewString()
+		group := uuid.NewString()
+		consumer1 := uuid.NewString()
+		consumer2 := uuid.NewString()
+
+		xadd, err := client.XAddWithOptions(
+			key,
+			[][]string{{"e1_f1", "e1_v1"}, {"e1_f2", "e1_v2"}},
+			*options.NewXAddOptions().SetId("0-1"),
+		)
+		assert.NoError(suite.T(), err)
+		assert.Equal(suite.T(), "0-1", xadd.Value())
+		xadd, err = client.XAddWithOptions(
+			key,
+			[][]string{{"e2_f1", "e2_v1"}, {"e2_f2", "e2_v2"}},
+			*options.NewXAddOptions().SetId("0-2"),
+		)
+		assert.NoError(suite.T(), err)
+		assert.Equal(suite.T(), "0-2", xadd.Value())
+		xadd, err = client.XAddWithOptions(key, [][]string{{"e3_f1", "e3_v1"}}, *options.NewXAddOptions().SetId("0-3"))
+		assert.NoError(suite.T(), err)
+		assert.Equal(suite.T(), "0-3", xadd.Value())
+
+		suite.verifyOK(client.XGroupCreate(key, group, "0-0"))
+
+		xReadGroup, err := client.XReadGroupWithOptions(
+			group,
+			consumer1,
+			map[string]string{key: ">"},
+			*options.NewXReadGroupOptions().SetCount(1),
+		)
+		assert.NoError(suite.T(), err)
+		expectedResult := map[string]map[string][][]string{
+			key: {
+				"0-1": {{"e1_f1", "e1_v1"}, {"e1_f2", "e1_v2"}},
+			},
+		}
+		assert.Equal(suite.T(), expectedResult, xReadGroup)
+
+		// Sleep to ensure the idle time value and inactive time value returned by xinfo_consumers is > 0
+		time.Sleep(2000 * time.Millisecond)
+		info, err := client.XInfoConsumers(key, group)
+		assert.NoError(suite.T(), err)
+		assert.Len(suite.T(), info, 1)
+		assert.Equal(suite.T(), consumer1, info[0].Name)
+		assert.Equal(suite.T(), int64(1), info[0].Pending)
+		assert.Greater(suite.T(), info[0].Idle, int64(0))
+		if suite.serverVersion > "7.2.0" {
+			assert.False(suite.T(), info[0].Inactive.IsNil())
+			assert.Greater(suite.T(), info[0].Inactive.Value(), int64(0))
+		} else {
+			assert.True(suite.T(), info[0].Inactive.IsNil())
+		}
+
+		respBool, err := client.XGroupCreateConsumer(key, group, consumer2)
+		assert.NoError(suite.T(), err)
+		assert.True(suite.T(), respBool)
+
+		xReadGroup, err = client.XReadGroup(group, consumer2, map[string]string{key: ">"})
+		assert.NoError(suite.T(), err)
+		expectedResult = map[string]map[string][][]string{
+			key: {
+				"0-2": {{"e2_f1", "e2_v1"}, {"e2_f2", "e2_v2"}},
+				"0-3": {{"e3_f1", "e3_v1"}},
+			},
+		}
+		assert.Equal(suite.T(), expectedResult, xReadGroup)
+
+		// Verify that xinfo_consumers contains info for 2 consumers now
+		info, err = client.XInfoConsumers(key, group)
+		assert.NoError(suite.T(), err)
+		assert.Len(suite.T(), info, 2)
+
+		// Passing a non-existing key raises an error
+		key = uuid.NewString()
+		_, err = client.XInfoConsumers(key, "_")
+		assert.IsType(suite.T(), &errors.RequestError{}, err)
+
+		// key exists, but it is not a stream
+		suite.verifyOK(client.Set(key, key))
+		_, err = client.XInfoConsumers(key, "_")
+		assert.IsType(suite.T(), &errors.RequestError{}, err)
+
+		// Passing a non-existing group raises an error
+		key = uuid.NewString()
+		_, err = client.XAdd(key, [][]string{{"a", "b"}})
+		assert.NoError(suite.T(), err)
+		_, err = client.XInfoConsumers(key, "_")
+		assert.IsType(suite.T(), &errors.RequestError{}, err)
+
+		// no consumers yet
+		suite.verifyOK(client.XGroupCreate(key, group, "0-0"))
+		info, err = client.XInfoConsumers(key, group)
+		assert.NoError(suite.T(), err)
+		assert.Empty(suite.T(), info)
+	})
+}
+
 func (suite *GlideTestSuite) TestXInfoGroups() {
 	suite.runWithDefaultClients(func(client api.BaseClient) {
 		key := uuid.NewString()