NOMKSTREAM support for XADD (redis#1507)

redis/client.py

@@ -2536,15 +2536,16 @@ def xack(self, name, groupname, *ids):
         """
         return self.execute_command('XACK', name, groupname, *ids)
 
-    def xadd(self, name, fields, id='*', maxlen=None, approximate=True):
+    def xadd(self, name, fields, id='*', maxlen=None, approximate=True,
+             nomkstream=False):
         """
         Add to a stream.
         name: name of the stream
         fields: dict of field/value pairs to insert into the stream
         id: Location to insert this record. By default it is appended.
         maxlen: truncate old stream members beyond this size
         approximate: actual stream length may be slightly more than maxlen
-
+        nomkstream: When set to true, do not make a stream
         """
         pieces = []
         if maxlen is not None:
@@ -2554,6 +2555,8 @@ def xadd(self, name, fields, id='*', maxlen=None, approximate=True):
             if approximate:
                 pieces.append(b'~')
             pieces.append(str(maxlen))
+        if nomkstream:
+            pieces.append(b'NOMKSTREAM')
         pieces.append(id)
         if not isinstance(fields, dict) or len(fields) == 0:
             raise DataError('XADD fields must be a non-empty dict')

tests/test_commands.py

@@ -2255,6 +2255,16 @@ def test_xadd(self, r):
         r.xadd(stream, {'foo': 'bar'}, maxlen=2, approximate=False)
         assert r.xlen(stream) == 2
 
+    @skip_if_server_version_lt('6.2.0')
+    def test_xadd_nomkstream(self, r):
+        # nomkstream option
+        stream = 'stream'
+        r.xadd(stream, {'foo': 'bar'})
+        r.xadd(stream, {'some': 'other'}, nomkstream=False)
+        assert r.xlen(stream) == 2
+        r.xadd(stream, {'some': 'other'}, nomkstream=True)
+        assert r.xlen(stream) == 3
+
     @skip_if_server_version_lt('5.0.0')
     def test_xclaim(self, r):
         stream = 'stream'