//XGROUP CREATE mystream mygroup 0，从流头开始
//只关心实时消息：用 XGROUP CREATE mystream mygroup $（推荐默认做法）
//已有 Group 想让新消费者补读：必须用 XREADGROUP GROUP mygroup newconsumer COUNT 100 STREAMS mystream 0，显式指定起始 ID 0；但注意这会干扰 Group 整体的 pending 状态管理

  //创建消费者组
  var lCmd3: IRedisCommand := NewRedisCommand('XGROUP');
  lCmd3
    .Add('create')
    .Add('mystream')   //流名
    .Add('mygroup')    //组名
    .Add('$')
    .Add('MKSTREAM');
  var lRes3: TRedisNullable<string> := fRedis.ExecuteWithStringResult(lCmd3);

  //为消费者组创建消费者
  var lCmd1: IRedisCommand := NewRedisCommand('XGROUP');
  lCmd1
    .Add('CREATECONSUMER')
    .Add('mystream')    //流名
    .Add('mygroup')     //组名
    .Add('myconsumer'); //消费者名
  var lRes1: TRedisNullable<string> := fRedis.ExecuteWithStringResult(lCmd1);

  //往流里插入一条消息
  var lCmd: IRedisCommand := NewRedisCommand('XADD');
  lCmd
    .Add('mystream')
    .Add('MAXLEN')
    .Add('~')
    .Add(10)
    .Add('*')
    .Add('key1')     //键1
    .Add('value1');  //值1
  var lRes: TRedisNullable<string> := fRedis.ExecuteWithStringResult(lCmd);
  Log(lRes.Value);

procedure TMainForm.Button2Click(Sender: TObject);
//订阅消费者组
begin
  TTask.Run(
    procedure
    begin
      while TTask.CurrentTask.Status <> TTaskStatus.Canceled do
      begin
        var lRedis := NewRedisClient();
        var lCmd: IRedisCommand := NewRedisCommand('XREADGROUP');
        lCmd
          .Add('group')
          .Add('mygroup')
          .Add('myconsumer')
          .Add('block')
          .Add(5000)
          .Add('count')
          .Add(1)
          .Add('streams')
          .Add('mystream')
          .Add('>');
        var lRes: TRedisRESPArray := lRedis.ExecuteAndGetRESPArray(lCmd);
        if Assigned(lRes) then
        Log(lRes.tostring);
      end;
    end);
end;

        var lRes: TRedisRESPArray := lRedis.ExecuteAndGetRESPArray(lCmd);
        if not Assigned(lRes) then
        begin
          Continue;
        end;
        try
          if lRes.Count > 0 then
          begin
           // var
            lSizeOfMyStreamArray := lRes
              .Items[0].ArrayValue
              .Items[1].ArrayValue
              .Count;

            lLastID := lRes
              .Items[0].ArrayValue
              .Items[1].ArrayValue
              .Items[lSizeOfMyStreamArray - 1].ArrayValue
              .Items[0].Value;
          end;
          Log(lRes.ToString+' count:'+lSizeOfMyStreamArray.ToString+' lastid:'+lLastID);
        finally
          lRes.Free;
        end;

uses
  redis.Command, redis.Client;

var
  Redis: IRedisClient;
  Cmd: IRedisCommand;
  AckCount: Integer;
begin
  Redis := NewRedisClient('127.0.0.1', 6379);
  
  // 方式一：直接构建命令（推荐，兼容性好）
  Cmd := NewRedisCommand('XACK')
    .Add('mystream')       // Stream Key
    .Add('mygroup')        // Consumer Group
    .Add('1526569495631-0'); // 消息 ID（可追加多个）
  
  AckCount := Redis.ExecuteWithIntegerResult(Cmd); 
  // 结果：成功确认的消息数
  
  // 方式二：批量确认多个 ID
  Cmd := NewRedisCommand('XACK')
    .Add('mystream')
    .Add('mygroup')
    .Add('1526569495631-0')
    .Add('1526569495632-0');
  AckCount := Redis.ExecuteWithIntegerResult(Cmd);
end