Method XREADGROUP

streams.go:689–735

(key string, group string, consumer string, option XReadOption, maxCount int64)

}

func (c Client) XREADGROUP(key string, group string, consumer string, option XReadOption, maxCount int64) (items []StreamItem, err error) {
	if option == XReadPending {
		return c.xGroupReadPending(key, group, consumer, maxCount)
	}

	retryCount := 0

	for retryCount < 5 {
		currentCursor, err := c.xGroupCursorGet(key, group)
		if err != nil {
			return items, err
		}

		items, err := c.XRANGE(key, currentCursor.Next(), XEnd, 1)

		if err != nil || len(items) == 0 {
			return items, err
		}

		item := items[0]

		var actions []dynamodb.TransactWriteItem
		actions = append(actions, c.xGroupCursorPushAction(key, group, item.ID))

		if option == XReadNew {
			actions = append(actions, PendingItem{
				ID:            item.ID,
				Consumer:      consumer,
				LastDelivered: time.Now(),
			}.toPutAction(c.xGroupKey(key, group), c))
		}

		_, err = c.ddbClient.TransactWriteItemsRequest(&dynamodb.TransactWriteItemsInput{
			TransactItems: actions,
		}).Send(context.TODO())
		if err == nil {
			return items, nil
		}

		if !conditionFailureError(err) {
			return items, err
		}
		retryCount++
	}

	return items, errors.New("too much contention")
}

// XREVRANGE is similar to XRANGE, but in reverse order. The stream items in descending chronological order. Using the
// same example as XRANGE, when fetching items in reverse order there are some differences when paginating. The first