XADD

"""
Code samples for XADD command:
    https://redis.io/docs/latest/commands/xadd/
"""

import redis

r = redis.Redis(decode_responses=True)

res1 = r.xadd("mystream", {"name": "Sara", "surname": "OConnor"})
print(res1)  # >>> 1726055713866-0

res2 = r.xadd("mystream", {"field1": "value1", "field2": "value2", "field3": "value3"})
print(res2)  # >>> 1726055713866-1

res3 = r.xlen("mystream")
print(res3)  # >>> 2

res4 = r.xrange("mystream", "-", "+")
print(res4)
# >>> [
#   ('1726055713866-0', {'name': 'Sara', 'surname': 'OConnor'}),
#   ('1726055713866-1', {'field1': 'value1', 'field2': 'value2', 'field3': 'value3'})
# ]


res5 = r.xadd("mystream", {"field": "value"}, idmp=("producer1", b"msg1"))
print(res5)  # >>> 1726055713867-0

# Attempting to add the same message again with IDMP returns the original entry ID
res6 = r.xadd("mystream", {"field": "different_value"}, idmp=("producer1", b"msg1"))
print(res6)  # >>> 1726055713867-0 (same ID as res5, message was deduplicated)

res7 = r.xadd("mystream", {"field": "value"}, idmpauto="producer2")
print(res7)  # >>> 1726055713867-1

# Auto-generated idempotent ID prevents duplicates for same producer+content
res8 = r.xadd("mystream", {"field": "value"}, idmpauto="producer2")
print(res8)  # >>> 1726055713867-1 (same ID as res7, duplicate detected)

# Configure idempotent message processing settings
res9 = r.xcfgset("mystream", idmp_duration=300, idmp_maxsize=1000)
print(res9)  # >>> True

import assert from 'node:assert';
import { createClient } from 'redis';

const client = createClient();
await client.connect();

const res1 = await client.xAdd('mystream', '*', {
  'name': 'Sara',
  'surname': 'OConnor'
});
console.log(res1); // >>> 1726055713866-0

const res2 = await client.xAdd('mystream', '*', {
  'field1': 'value1',
  'field2': 'value2',
  'field3': 'value3'
});
console.log(res2); // >>> 1726055713866-1

const res3 = await client.xLen('mystream');
console.log(res3); // >>> 2

const res4 = await client.xRange('mystream', '-', '+');
console.log(res4);
// >>> [
//   { id: '1726055713866-0', message: { name: 'Sara', surname: 'OConnor' } },
//   { id: '1726055713866-1', message: { field1: 'value1', field2: 'value2', field3: 'value3' } }
// ]


const res5 = await client.xAdd('mystream', '*', { 'field': 'value' }, {
  IDMP: { pid: 'producer1', iid: 'msg1' }
});
console.log(res5); // >>> 1726055713867-0

// Attempting to add the same message again with IDMP returns the original entry ID
const res6 = await client.xAdd('mystream', '*', { 'field': 'different_value' }, {
  IDMP: { pid: 'producer1', iid: 'msg1' }
});
console.log(res6); // >>> 1726055713867-0 (same ID as res5, message was deduplicated)

const res7 = await client.xAdd('mystream', '*', { 'field': 'value' }, {
  IDMPAUTO: { pid: 'producer2' }
});
console.log(res7); // >>> 1726055713867-1

// Auto-generated idempotent ID prevents duplicates for same producer+content
const res8 = await client.xAdd('mystream', '*', { 'field': 'value' }, {
  IDMPAUTO: { pid: 'producer2' }
});
console.log(res8); // >>> 1726055713867-1 (same ID as res7, duplicate detected)

// Configure idempotent message processing settings
const res9 = await client.xCfgSet('mystream', {
  IDMP_DURATION: 300,
  IDMP_MAXSIZE: 1000
});
console.log(res9); // >>> OK


await client.quit();

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import redis.clients.jedis.JedisPooled;
import redis.clients.jedis.StreamEntryID;
import redis.clients.jedis.params.XAddParams;
import redis.clients.jedis.params.XCfgSetParams;
import redis.clients.jedis.resps.StreamEntry;

import static org.junit.jupiter.api.Assertions.*;

public class CmdsStreamExample {

    public void run() {
        JedisPooled jedis = new JedisPooled("localhost", 6379);


        Map<String, String> entry1 = new HashMap<>();
        entry1.put("name", "Sara");
        entry1.put("surname", "OConnor");
        StreamEntryID res1 = jedis.xadd("mystream", entry1, XAddParams.xAddParams());
        System.out.println(res1); // >>> 1726055713866-0

        Map<String, String> entry2 = new HashMap<>();
        entry2.put("field1", "value1");
        entry2.put("field2", "value2");
        entry2.put("field3", "value3");
        StreamEntryID res2 = jedis.xadd("mystream", entry2, XAddParams.xAddParams());
        System.out.println(res2); // >>> 1726055713866-1

        long res3 = jedis.xlen("mystream");
        System.out.println(res3); // >>> 2

        List<StreamEntry> res4 = jedis.xrange("mystream", "-", "+");
        for (StreamEntry entry : res4) {
            System.out.println(entry.getID() + " -> " + entry.getFields());
        }
        // >>> 1726055713866-0 -> {name=Sara, surname=OConnor}
        // >>> 1726055713866-1 -> {field1=value1, field2=value2, field3=value3}

        Map<String, String> idmpEntry1 = new HashMap<>();
        idmpEntry1.put("field", "value");
        StreamEntryID res5 = jedis.xadd("mystream", idmpEntry1, 
            XAddParams.xAddParams().idmp("producer1", "msg1"));
        System.out.println(res5); // >>> 1726055713867-0

        // Attempting to add the same message again with IDMP returns the original entry ID
        Map<String, String> idmpEntry2 = new HashMap<>();
        idmpEntry2.put("field", "different_value");
        StreamEntryID res6 = jedis.xadd("mystream", idmpEntry2,
            XAddParams.xAddParams().idmp("producer1", "msg1"));
        System.out.println(res6); // >>> 1726055713867-0 (same ID as res5, message was deduplicated)

        Map<String, String> idmpAutoEntry1 = new HashMap<>();
        idmpAutoEntry1.put("field", "value");
        StreamEntryID res7 = jedis.xadd("mystream", idmpAutoEntry1,
            XAddParams.xAddParams().idmpAuto("producer2"));
        System.out.println(res7); // >>> 1726055713867-1

        // Auto-generated idempotent ID prevents duplicates for same producer+content
        Map<String, String> idmpAutoEntry2 = new HashMap<>();
        idmpAutoEntry2.put("field", "value");
        StreamEntryID res8 = jedis.xadd("mystream", idmpAutoEntry2,
            XAddParams.xAddParams().idmpAuto("producer2"));
        System.out.println(res8); // >>> 1726055713867-1 (same ID as res7, duplicate detected)

        // Configure idempotent message processing settings
        String res9 = jedis.xcfgset("mystream",
            XCfgSetParams.xCfgSetParams().idmpDuration(300).idmpMaxsize(1000));
        System.out.println(res9); // >>> OK

        jedis.close();
    }
}

package io.redis.examples.async;

import io.lettuce.core.*;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.StatefulRedisConnection;

import java.util.*;
import java.util.concurrent.CompletableFuture;

public class CmdsStreamExample {

    public void run() {
        RedisClient redisClient = RedisClient.create("redis://localhost:6379");

        try (StatefulRedisConnection<String, String> connection = redisClient.connect()) {
            RedisAsyncCommands<String, String> asyncCommands = connection.async();


            Map<String, String> entry1 = new HashMap<>();
            entry1.put("name", "Sara");
            entry1.put("surname", "OConnor");

            CompletableFuture<Void> xadd1Ops = asyncCommands.xadd("mystream", entry1)
                .thenCompose(res1 -> {
                    System.out.println(res1); // >>> 1726055713866-0
                    Map<String, String> entry2 = new HashMap<>();
                    entry2.put("field1", "value1");
                    entry2.put("field2", "value2");
                    entry2.put("field3", "value3");
                    return asyncCommands.xadd("mystream", entry2);
                })
                .thenCompose(res2 -> {
                    System.out.println(res2); // >>> 1726055713866-1
                    return asyncCommands.xlen("mystream");
                })
                .thenCompose(res3 -> {
                    System.out.println(res3); // >>> 2
                    return asyncCommands.xrange("mystream", Range.unbounded());
                })
                .thenAccept(res4 -> {
                    for (var entry : res4) {
                        System.out.println(entry.getId() + " -> " + entry.getBody());
                    }
                    // >>> 1726055713866-0 -> {name=Sara, surname=OConnor}
                    // >>> 1726055713866-1 -> {field1=value1, field2=value2, field3=value3}
                })
                .toCompletableFuture();

            xadd1Ops.join();


            Map<String, String> idmpEntry1 = new HashMap<>();
            idmpEntry1.put("field", "value");

            CompletableFuture<Void> xadd2Ops = asyncCommands.xadd(
                    "mystream",
                    XAddArgs.Builder.idmp("producer1", "msg1"),
                    idmpEntry1)
                .thenCompose(res5 -> {
                    System.out.println(res5); // >>> 1726055713867-0
                    // Attempting to add same message again with IDMP returns original entry ID
                    Map<String, String> idmpEntry2 = new HashMap<>();
                    idmpEntry2.put("field", "different_value");
                    return asyncCommands.xadd("mystream",
                        XAddArgs.Builder.idmp("producer1", "msg1"), idmpEntry2);
                })
                .thenCompose(res6 -> {
                    System.out.println(res6); // >>> 1726055713867-0 (deduplicated)
                    Map<String, String> idmpAutoEntry1 = new HashMap<>();
                    idmpAutoEntry1.put("field", "value");
                    return asyncCommands.xadd("mystream",
                        XAddArgs.Builder.idmpAuto("producer2"), idmpAutoEntry1);
                })
                .thenCompose(res7 -> {
                    System.out.println(res7); // >>> 1726055713867-1
                    Map<String, String> idmpAutoEntry2 = new HashMap<>();
                    idmpAutoEntry2.put("field", "value");
                    return asyncCommands.xadd("mystream",
                        XAddArgs.Builder.idmpAuto("producer2"), idmpAutoEntry2);
                })
                .thenAccept(res8 -> {
                    System.out.println(res8); // >>> 1726055713867-1 (duplicate detected)
                })
                .toCompletableFuture();

            xadd2Ops.join();


        } finally {
            redisClient.shutdown();
        }
    }
}

package io.redis.examples.reactive;

import io.lettuce.core.*;
import io.lettuce.core.api.reactive.RedisReactiveCommands;
import io.lettuce.core.api.StatefulRedisConnection;


import reactor.core.publisher.Mono;

import java.util.*;

public class CmdsStreamExample {

    public void run() {
        RedisClient redisClient = RedisClient.create("redis://localhost:6379");

        try (StatefulRedisConnection<String, String> connection = redisClient.connect()) {
            RedisReactiveCommands<String, String> reactiveCommands = connection.reactive();


            Map<String, String> entry1 = new HashMap<>();
            entry1.put("name", "Sara");
            entry1.put("surname", "OConnor");

            Mono<String> addEntry1 = reactiveCommands.xadd("mystream", entry1)
                .doOnNext(res1 -> {
                    System.out.println(res1); // >>> 1726055713866-0
                });

            addEntry1.block();

            Map<String, String> entry2 = new HashMap<>();
            entry2.put("field1", "value1");
            entry2.put("field2", "value2");
            entry2.put("field3", "value3");

            Mono<String> addEntry2 = reactiveCommands.xadd("mystream", entry2)
                .doOnNext(res2 -> {
                    System.out.println(res2); // >>> 1726055713866-1
                });

            addEntry2.block();

            Mono<Long> getLen = reactiveCommands.xlen("mystream")
                .doOnNext(res3 -> {
                    System.out.println(res3); // >>> 2
                });

            getLen.block();

            Mono<List<StreamMessage<String, String>>> getRange = reactiveCommands
                .xrange("mystream", Range.unbounded())
                .collectList()
                .doOnNext(res4 -> {
                    for (var entry : res4) {
                        System.out.println(entry.getId() + " -> " + entry.getBody());
                    }
                    // >>> 1726055713866-0 -> {name=Sara, surname=OConnor}
                    // >>> 1726055713866-1 -> {field1=value1, field2=value2, field3=value3}
                });

            getRange.block();


            Map<String, String> idmpEntry1 = new HashMap<>();
            idmpEntry1.put("field", "value");

            Mono<String> addIdmp1 = reactiveCommands.xadd(
                    "mystream",
                    XAddArgs.Builder.idmp("producer1", "msg1"),
                    idmpEntry1)
                .doOnNext(res5 -> {
                    System.out.println(res5); // >>> 1726055713867-0
                });

            addIdmp1.block();

            // Attempting to add the same message again with IDMP returns the original entry ID
            Map<String, String> idmpEntry2 = new HashMap<>();
            idmpEntry2.put("field", "different_value");

            Mono<String> addIdmp2 = reactiveCommands.xadd(
                    "mystream",
                    XAddArgs.Builder.idmp("producer1", "msg1"),
                    idmpEntry2)
                .doOnNext(res6 -> {
                    System.out.println(res6); // >>> 1726055713867-0 (deduplicated)
                });

            addIdmp2.block();

            Map<String, String> idmpAutoEntry1 = new HashMap<>();
            idmpAutoEntry1.put("field", "value");

            Mono<String> addIdmpAuto1 = reactiveCommands.xadd(
                    "mystream",
                    XAddArgs.Builder.idmpAuto("producer2"),
                    idmpAutoEntry1)
                .doOnNext(res7 -> {
                    System.out.println(res7); // >>> 1726055713867-1
                });

            addIdmpAuto1.block();

            // Auto-generated idempotent ID prevents duplicates for same producer+content
            Map<String, String> idmpAutoEntry2 = new HashMap<>();
            idmpAutoEntry2.put("field", "value");

            Mono<String> addIdmpAuto2 = reactiveCommands.xadd(
                    "mystream",
                    XAddArgs.Builder.idmpAuto("producer2"),
                    idmpAutoEntry2)
                .doOnNext(res8 -> {
                    System.out.println(res8); // >>> 1726055713867-1 (duplicate detected)
                });

            addIdmpAuto2.block();


        } finally {
            redisClient.shutdown();
        }
    }
}

package example_commands_test

import (
	"context"
	"fmt"

	"github.com/redis/go-redis/v9"
)


func ExampleClient_xadd1() {
	ctx := context.Background()

	rdb := redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "",
		DB:       0,
	})


	res1, err := rdb.XAdd(ctx, &redis.XAddArgs{
		Stream: "mystream",
		Values: map[string]interface{}{
			"name":    "Sara",
			"surname": "OConnor",
		},
	}).Result()

	if err != nil {
		panic(err)
	}

	fmt.Println(res1 != "") // >>> true

	res2, err := rdb.XAdd(ctx, &redis.XAddArgs{
		Stream: "mystream",
		Values: map[string]interface{}{
			"field1": "value1",
			"field2": "value2",
			"field3": "value3",
		},
	}).Result()

	if err != nil {
		panic(err)
	}

	fmt.Println(res2 != "") // >>> true

	res3, err := rdb.XLen(ctx, "mystream").Result()

	if err != nil {
		panic(err)
	}

	fmt.Println(res3) // >>> 2

	res4, err := rdb.XRange(ctx, "mystream", "-", "+").Result()

	if err != nil {
		panic(err)
	}

	fmt.Println(len(res4)) // >>> 2

}

func ExampleClient_xadd2() {
	ctx := context.Background()

	rdb := redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "",
		DB:       0,
	})


	res5, err := rdb.XAdd(ctx, &redis.XAddArgs{
		Stream:       "mystream",
		Values:       map[string]interface{}{"field": "value"},
		ProducerID:   "producer1",
		IdempotentID: "msg1",
	}).Result()

	if err != nil {
		panic(err)
	}

	fmt.Println(res5 != "") // >>> true

	// Attempting to add the same message again with IDMP returns the original entry ID
	res6, err := rdb.XAdd(ctx, &redis.XAddArgs{
		Stream:       "mystream",
		Values:       map[string]interface{}{"field": "different_value"},
		ProducerID:   "producer1",
		IdempotentID: "msg1",
	}).Result()

	if err != nil {
		panic(err)
	}

	fmt.Println(res5 == res6) // >>> true (same ID, message was deduplicated)

	res7, err := rdb.XAdd(ctx, &redis.XAddArgs{
		Stream:         "mystream",
		Values:         map[string]interface{}{"field": "value"},
		ProducerID:     "producer2",
		IdempotentAuto: true,
	}).Result()

	if err != nil {
		panic(err)
	}

	fmt.Println(res7 != "") // >>> true

	// Auto-generated idempotent ID prevents duplicates for same producer+content
	res8, err := rdb.XAdd(ctx, &redis.XAddArgs{
		Stream:         "mystream",
		Values:         map[string]interface{}{"field": "value"},
		ProducerID:     "producer2",
		IdempotentAuto: true,
	}).Result()

	if err != nil {
		panic(err)
	}

	fmt.Println(res7 == res8) // >>> true (same ID, duplicate detected)

	// Configure idempotent message processing settings
	res9, err := rdb.XCfgSet(ctx, &redis.XCfgSetArgs{
		Stream:   "mystream",
		Duration: 300,  // 300 seconds
		MaxSize:  1000, // 1000 idempotent IDs
	}).Result()

	if err != nil {
		panic(err)
	}

	fmt.Println(res9) // >>> OK

}

using StackExchange.Redis;

public class CmdsStreamExample
{
    public void Run()
    {
        var muxer = ConnectionMultiplexer.Connect("localhost:6379");
        var db = muxer.GetDatabase();


        var res1 = db.StreamAdd("mystream", new NameValueEntry[] {
            new NameValueEntry("name", "Sara"),
            new NameValueEntry("surname", "OConnor")
        });
        Console.WriteLine(res1);  // >>> 1726055713866-0

        var res2 = db.StreamAdd("mystream", new NameValueEntry[] {
            new NameValueEntry("field1", "value1"),
            new NameValueEntry("field2", "value2"),
            new NameValueEntry("field3", "value3")
        });
        Console.WriteLine(res2);  // >>> 1726055713866-1

        var res3 = db.StreamLength("mystream");
        Console.WriteLine(res3);  // >>> 2

        var res4 = db.StreamRange("mystream", "-", "+");
        foreach (var entry in res4)
        {
            Console.WriteLine($"{entry.Id} -> {string.Join(", ", entry.Values.Select(v => $"{v.Name}={v.Value}"))}");
        }
        // >>> 1726055713866-0 -> name=Sara, surname=OConnor
        // >>> 1726055713866-1 -> field1=value1, field2=value2, field3=value3


        // Note: IDMP is a Redis 8.6 feature - using Execute for raw command access
        var res5 = db.Execute("XADD", "mystream", "IDMP", "producer1", "msg1", "*", "field", "value");
        Console.WriteLine(res5);  // >>> 1726055713867-0

        // Attempting to add the same message again with IDMP returns the original entry ID
        var res6 = db.Execute("XADD", "mystream", "IDMP", "producer1", "msg1", "*", "field", "different_value");
        Console.WriteLine(res6);  // >>> 1726055713867-0 (same ID as res5, message was deduplicated)

        var res7 = db.Execute("XADD", "mystream", "IDMPAUTO", "producer2", "*", "field", "value");
        Console.WriteLine(res7);  // >>> 1726055713867-1

        // Auto-generated idempotent ID prevents duplicates for same producer+content
        var res8 = db.Execute("XADD", "mystream", "IDMPAUTO", "producer2", "*", "field", "value");
        Console.WriteLine(res8);  // >>> 1726055713867-1 (same ID as res7, duplicate detected)

        // Configure idempotent message processing settings
        var res9 = db.Execute("XCFGSET", "mystream", "IDMP-DURATION", "300", "IDMP-MAXSIZE", "1000");
        Console.WriteLine(res9);  // >>> OK


        muxer.Close();
    }
}

<?php

use PHPUnit\Framework\TestCase;
use Predis\Client as PredisClient;

class CmdStreamTest
{
    public function testXadd1(): void
    {
        $redis = new PredisClient([
            'scheme' => 'tcp',
            'host'   => '127.0.0.1',
            'port'   => 6379,
        ]);

        $res1 = $redis->xadd('mystream', ['name' => 'Sara', 'surname' => 'OConnor']);
        echo $res1 . PHP_EOL; // >>> 1726055713866-0

        $res2 = $redis->xadd('mystream', ['field1' => 'value1', 'field2' => 'value2', 'field3' => 'value3']);
        echo $res2 . PHP_EOL; // >>> 1726055713866-1

        $res3 = $redis->xlen('mystream');
        echo $res3 . PHP_EOL; // >>> 2

        $res4 = $redis->xrange('mystream', '-', '+');
        foreach ($res4 as $id => $fields) {
            echo $id . ' -> ' . json_encode($fields) . PHP_EOL;
        }
        // >>> 1726055713866-0 -> {"name":"Sara","surname":"OConnor"}
        // >>> 1726055713866-1 -> {"field1":"value1","field2":"value2","field3":"value3"}

    }

    public function testXadd2(): void
    {
        $redis = new PredisClient([
            'scheme' => 'tcp',
            'host'   => '127.0.0.1',
            'port'   => 6379,
        ]);

        $res5 = $redis->xadd('mystream', ['field' => 'value'], '*', ['idmp' => ['producer1', 'msg1']]);
        echo $res5 . PHP_EOL; // >>> 1726055713867-0

        // Attempting to add the same message again with IDMP returns the original entry ID
        $res6 = $redis->xadd('mystream', ['field' => 'different_value'], '*', ['idmp' => ['producer1', 'msg1']]);
        echo $res6 . PHP_EOL; // >>> 1726055713867-0 (same ID as res5, message was deduplicated)

        $res7 = $redis->xadd('mystream', ['field' => 'value'], '*', ['idmpauto' => 'producer2']);
        echo $res7 . PHP_EOL; // >>> 1726055713867-1

        // Auto-generated idempotent ID prevents duplicates for same producer+content
        $res8 = $redis->xadd('mystream', ['field' => 'value'], '*', ['idmpauto' => 'producer2']);
        echo $res8 . PHP_EOL; // >>> 1726055713867-1 (same ID as res7, duplicate detected)

        // Configure idempotent message processing settings
        $res9 = $redis->xcfgset('mystream', 300, 1000);
        echo $res9 . PHP_EOL; // >>> OK

    }
}