We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add the "MKSTREAM" option to xgroup_create
It's a not-yet-documented option for creating consumer groups before the stream exists: https://github.com/antirez/redis/issues/4824#issuecomment-438263421

Example signature for `StreamCommandsMixin`:
```py
    def xgroup_create(self, stream, group_name, latest_id='$', mkstream=False):
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `aioredis/commands/streams.py`
Content:
```
1 from collections import OrderedDict
2 
3 from aioredis.util import wait_convert, wait_make_dict, wait_ok
4 
5 
6 def fields_to_dict(fields, type_=OrderedDict):
7     """Convert a flat list of key/values into an OrderedDict"""
8     fields_iterator = iter(fields)
9     return type_(zip(fields_iterator, fields_iterator))
10 
11 
12 def parse_messages(messages):
13     """ Parse messages as returned by Redis into something useful
14 
15     Messages returned by XRANGE arrive in the form:
16 
17         [
18             [message_id, [key1, value1, key2, value2, ...]],
19             ...
20         ]
21 
22     Here we parse this into:
23 
24         [
25             [message_id, OrderedDict(
26                 (key1, value1),
27                 (key2, value2),
28                 ...
29             )],
30             ...
31         ]
32 
33     """
34     if messages is None:
35         return []
36     return [(mid, fields_to_dict(values)) for mid, values in messages]
37 
38 
39 def parse_messages_by_stream(messages_by_stream):
40     """ Parse messages returned by stream
41 
42     Messages returned by XREAD arrive in the form:
43         [stream_name,
44             [
45                 [message_id, [key1, value1, key2, value2, ...]],
46                 ...
47             ],
48             ...
49         ]
50 
51     Here we parse this into (with the help of the above parse_messages()
52     function):
53 
54         [
55             [stream_name, message_id, OrderedDict(
56                 (key1, value1),
57                 (key2, value2),.
58                 ...
59             )],
60             ...
61         ]
62 
63     """
64     if messages_by_stream is None:
65         return []
66 
67     parsed = []
68     for stream, messages in messages_by_stream:
69         for message_id, fields in parse_messages(messages):
70             parsed.append((stream, message_id, fields))
71     return parsed
72 
73 
74 def parse_lists_to_dicts(lists):
75     """ Convert [[a, 1, b, 2], ...] into [{a:1, b: 2}, ...]"""
76     return [fields_to_dict(l, type_=dict) for l in lists]
77 
78 
79 class StreamCommandsMixin:
80     """Stream commands mixin
81 
82     Streams are under development in Redis and
83     not currently released.
84     """
85 
86     def xadd(self, stream, fields, message_id=b'*', max_len=None,
87              exact_len=False):
88         """Add a message to a stream."""
89         args = []
90         if max_len is not None:
91             if exact_len:
92                 args.extend((b'MAXLEN', max_len))
93             else:
94                 args.extend((b'MAXLEN', b'~', max_len))
95 
96         args.append(message_id)
97 
98         for k, v in fields.items():
99             args.extend([k, v])
100         return self.execute(b'XADD', stream, *args)
101 
102     def xrange(self, stream, start='-', stop='+', count=None):
103         """Retrieve messages from a stream."""
104         if count is not None:
105             extra = ['COUNT', count]
106         else:
107             extra = []
108         fut = self.execute(b'XRANGE', stream, start, stop, *extra)
109         return wait_convert(fut, parse_messages)
110 
111     def xrevrange(self, stream, start='+', stop='-', count=None):
112         """Retrieve messages from a stream in reverse order."""
113         if count is not None:
114             extra = ['COUNT', count]
115         else:
116             extra = []
117         fut = self.execute(b'XREVRANGE', stream, start, stop, *extra)
118         return wait_convert(fut, parse_messages)
119 
120     def xread(self, streams, timeout=0, count=None, latest_ids=None):
121         """Perform a blocking read on the given stream
122 
123         :raises ValueError: if the length of streams and latest_ids do
124                             not match
125         """
126         args = self._xread(streams, timeout, count, latest_ids)
127         fut = self.execute(b'XREAD', *args)
128         return wait_convert(fut, parse_messages_by_stream)
129 
130     def xread_group(self, group_name, consumer_name, streams, timeout=0,
131                     count=None, latest_ids=None):
132         """Perform a blocking read on the given stream as part of a consumer group
133 
134         :raises ValueError: if the length of streams and latest_ids do
135                             not match
136         """
137         args = self._xread(streams, timeout, count, latest_ids)
138         fut = self.execute(
139             b'XREADGROUP', b'GROUP', group_name, consumer_name, *args
140         )
141         return wait_convert(fut, parse_messages_by_stream)
142 
143     def xgroup_create(self, stream, group_name, latest_id='$'):
144         """Create a consumer group"""
145         fut = self.execute(b'XGROUP', b'CREATE', stream, group_name, latest_id)
146         return wait_ok(fut)
147 
148     def xgroup_setid(self, stream, group_name, latest_id='$'):
149         """Set the latest ID for a consumer group"""
150         fut = self.execute(b'XGROUP', b'SETID', stream, group_name, latest_id)
151         return wait_ok(fut)
152 
153     def xgroup_destroy(self, stream, group_name):
154         """Delete a consumer group"""
155         fut = self.execute(b'XGROUP', b'DESTROY', stream, group_name)
156         return wait_ok(fut)
157 
158     def xgroup_delconsumer(self, stream, group_name, consumer_name):
159         """Delete a specific consumer from a group"""
160         fut = self.execute(
161             b'XGROUP', b'DELCONSUMER', stream, group_name, consumer_name
162         )
163         return wait_convert(fut, int)
164 
165     def xpending(self, stream, group_name, start=None, stop=None, count=None,
166                  consumer=None):
167         """Get information on pending messages for a stream
168 
169         Returned data will vary depending on the presence (or not)
170         of the start/stop/count parameters. For more details see:
171         https://redis.io/commands/xpending
172 
173         :raises ValueError: if the start/stop/count parameters are only
174                             partially specified
175         """
176         # Returns: total pel messages, min id, max id, count
177         ssc = [start, stop, count]
178         ssc_count = len([v for v in ssc if v is not None])
179         if ssc_count != 3 and ssc_count != 0:
180             raise ValueError(
181                 'Either specify non or all of the start/stop/count arguments'
182             )
183         if not any(ssc):
184             ssc = []
185 
186         args = [stream, group_name] + ssc
187         if consumer:
188             args.append(consumer)
189         return self.execute(b'XPENDING', *args)
190 
191     def xclaim(self, stream, group_name, consumer_name, min_idle_time,
192                id, *ids):
193         """Claim a message for a given consumer"""
194         fut = self.execute(
195             b'XCLAIM', stream, group_name, consumer_name, min_idle_time,
196             id, *ids
197         )
198         return wait_convert(fut, parse_messages)
199 
200     def xack(self, stream, group_name, id, *ids):
201         """Acknowledge a message for a given consumer group"""
202         return self.execute(b'XACK', stream, group_name, id, *ids)
203 
204     def xinfo(self, stream):
205         """Retrieve information about the given stream.
206 
207         An alias for ``xinfo_stream()``
208         """
209         return self.xinfo_stream(stream)
210 
211     def xinfo_consumers(self, stream, group_name):
212         """Retrieve consumers of a consumer group"""
213         fut = self.execute(b'XINFO', b'CONSUMERS', stream, group_name)
214 
215         return wait_convert(fut, parse_lists_to_dicts)
216 
217     def xinfo_groups(self, stream):
218         """Retrieve the consumer groups for a stream"""
219         fut = self.execute(b'XINFO', b'GROUPS', stream)
220         return wait_convert(fut, parse_lists_to_dicts)
221 
222     def xinfo_stream(self, stream):
223         """Retrieve information about the given stream."""
224         fut = self.execute(b'XINFO', b'STREAM', stream)
225         return wait_make_dict(fut)
226 
227     def xinfo_help(self):
228         """Retrieve help regarding the ``XINFO`` sub-commands"""
229         fut = self.execute(b'XINFO', b'HELP')
230         return wait_convert(fut, lambda l: b'\n'.join(l))
231 
232     def _xread(self, streams, timeout=0, count=None, latest_ids=None):
233         """Wraps up common functionality between ``xread()``
234         and ``xread_group()``
235 
236         You should probably be using ``xread()`` or ``xread_group()`` directly.
237         """
238         if latest_ids is None:
239             latest_ids = ['$'] * len(streams)
240         if len(streams) != len(latest_ids):
241             raise ValueError(
242                 'The streams and latest_ids parameters must be of the '
243                 'same length'
244             )
245 
246         count_args = [b'COUNT', count] if count else []
247         if timeout is None:
248             block_args = []
249         else:
250             block_args = [b'BLOCK', timeout]
251         return block_args + count_args + [b'STREAMS'] + streams + latest_ids
252
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/aioredis/commands/streams.py b/aioredis/commands/streams.py
--- a/aioredis/commands/streams.py
+++ b/aioredis/commands/streams.py
@@ -79,8 +79,7 @@
 class StreamCommandsMixin:
     """Stream commands mixin
 
-    Streams are under development in Redis and
-    not currently released.
+    Streams are available in Redis since v5.0
     """
 
     def xadd(self, stream, fields, message_id=b'*', max_len=None,
@@ -140,9 +139,12 @@
         )
         return wait_convert(fut, parse_messages_by_stream)
 
-    def xgroup_create(self, stream, group_name, latest_id='$'):
+    def xgroup_create(self, stream, group_name, latest_id='$', mkstream=False):
         """Create a consumer group"""
-        fut = self.execute(b'XGROUP', b'CREATE', stream, group_name, latest_id)
+        args = [b'CREATE', stream, group_name, latest_id]
+        if mkstream:
+            args.append(b'MKSTREAM')
+        fut = self.execute(b'XGROUP', *args)
         return wait_ok(fut)
 
     def xgroup_setid(self, stream, group_name, latest_id='$'):

{"golden_diff": "diff --git a/aioredis/commands/streams.py b/aioredis/commands/streams.py\n--- a/aioredis/commands/streams.py\n+++ b/aioredis/commands/streams.py\n@@ -79,8 +79,7 @@\n class StreamCommandsMixin:\n     \"\"\"Stream commands mixin\n \n-    Streams are under development in Redis and\n-    not currently released.\n+    Streams are available in Redis since v5.0\n     \"\"\"\n \n     def xadd(self, stream, fields, message_id=b'*', max_len=None,\n@@ -140,9 +139,12 @@\n         )\n         return wait_convert(fut, parse_messages_by_stream)\n \n-    def xgroup_create(self, stream, group_name, latest_id='$'):\n+    def xgroup_create(self, stream, group_name, latest_id='$', mkstream=False):\n         \"\"\"Create a consumer group\"\"\"\n-        fut = self.execute(b'XGROUP', b'CREATE', stream, group_name, latest_id)\n+        args = [b'CREATE', stream, group_name, latest_id]\n+        if mkstream:\n+            args.append(b'MKSTREAM')\n+        fut = self.execute(b'XGROUP', *args)\n         return wait_ok(fut)\n \n     def xgroup_setid(self, stream, group_name, latest_id='$'):\n", "issue": "Add the \"MKSTREAM\" option to xgroup_create\nIt's a not-yet-documented option for creating consumer groups before the stream exists: https://github.com/antirez/redis/issues/4824#issuecomment-438263421\r\n\r\nExample signature for `StreamCommandsMixin`:\r\n```py\r\n    def xgroup_create(self, stream, group_name, latest_id='$', mkstream=False):\r\n```\n", "before_files": [{"content": "from collections import OrderedDict\n\nfrom aioredis.util import wait_convert, wait_make_dict, wait_ok\n\n\ndef fields_to_dict(fields, type_=OrderedDict):\n    \"\"\"Convert a flat list of key/values into an OrderedDict\"\"\"\n    fields_iterator = iter(fields)\n    return type_(zip(fields_iterator, fields_iterator))\n\n\ndef parse_messages(messages):\n    \"\"\" Parse messages as returned by Redis into something useful\n\n    Messages returned by XRANGE arrive in the form:\n\n        [\n            [message_id, [key1, value1, key2, value2, ...]],\n            ...\n        ]\n\n    Here we parse this into:\n\n        [\n            [message_id, OrderedDict(\n                (key1, value1),\n                (key2, value2),\n                ...\n            )],\n            ...\n        ]\n\n    \"\"\"\n    if messages is None:\n        return []\n    return [(mid, fields_to_dict(values)) for mid, values in messages]\n\n\ndef parse_messages_by_stream(messages_by_stream):\n    \"\"\" Parse messages returned by stream\n\n    Messages returned by XREAD arrive in the form:\n        [stream_name,\n            [\n                [message_id, [key1, value1, key2, value2, ...]],\n                ...\n            ],\n            ...\n        ]\n\n    Here we parse this into (with the help of the above parse_messages()\n    function):\n\n        [\n            [stream_name, message_id, OrderedDict(\n                (key1, value1),\n                (key2, value2),.\n                ...\n            )],\n            ...\n        ]\n\n    \"\"\"\n    if messages_by_stream is None:\n        return []\n\n    parsed = []\n    for stream, messages in messages_by_stream:\n        for message_id, fields in parse_messages(messages):\n            parsed.append((stream, message_id, fields))\n    return parsed\n\n\ndef parse_lists_to_dicts(lists):\n    \"\"\" Convert [[a, 1, b, 2], ...] into [{a:1, b: 2}, ...]\"\"\"\n    return [fields_to_dict(l, type_=dict) for l in lists]\n\n\nclass StreamCommandsMixin:\n    \"\"\"Stream commands mixin\n\n    Streams are under development in Redis and\n    not currently released.\n    \"\"\"\n\n    def xadd(self, stream, fields, message_id=b'*', max_len=None,\n             exact_len=False):\n        \"\"\"Add a message to a stream.\"\"\"\n        args = []\n        if max_len is not None:\n            if exact_len:\n                args.extend((b'MAXLEN', max_len))\n            else:\n                args.extend((b'MAXLEN', b'~', max_len))\n\n        args.append(message_id)\n\n        for k, v in fields.items():\n            args.extend([k, v])\n        return self.execute(b'XADD', stream, *args)\n\n    def xrange(self, stream, start='-', stop='+', count=None):\n        \"\"\"Retrieve messages from a stream.\"\"\"\n        if count is not None:\n            extra = ['COUNT', count]\n        else:\n            extra = []\n        fut = self.execute(b'XRANGE', stream, start, stop, *extra)\n        return wait_convert(fut, parse_messages)\n\n    def xrevrange(self, stream, start='+', stop='-', count=None):\n        \"\"\"Retrieve messages from a stream in reverse order.\"\"\"\n        if count is not None:\n            extra = ['COUNT', count]\n        else:\n            extra = []\n        fut = self.execute(b'XREVRANGE', stream, start, stop, *extra)\n        return wait_convert(fut, parse_messages)\n\n    def xread(self, streams, timeout=0, count=None, latest_ids=None):\n        \"\"\"Perform a blocking read on the given stream\n\n        :raises ValueError: if the length of streams and latest_ids do\n                            not match\n        \"\"\"\n        args = self._xread(streams, timeout, count, latest_ids)\n        fut = self.execute(b'XREAD', *args)\n        return wait_convert(fut, parse_messages_by_stream)\n\n    def xread_group(self, group_name, consumer_name, streams, timeout=0,\n                    count=None, latest_ids=None):\n        \"\"\"Perform a blocking read on the given stream as part of a consumer group\n\n        :raises ValueError: if the length of streams and latest_ids do\n                            not match\n        \"\"\"\n        args = self._xread(streams, timeout, count, latest_ids)\n        fut = self.execute(\n            b'XREADGROUP', b'GROUP', group_name, consumer_name, *args\n        )\n        return wait_convert(fut, parse_messages_by_stream)\n\n    def xgroup_create(self, stream, group_name, latest_id='$'):\n        \"\"\"Create a consumer group\"\"\"\n        fut = self.execute(b'XGROUP', b'CREATE', stream, group_name, latest_id)\n        return wait_ok(fut)\n\n    def xgroup_setid(self, stream, group_name, latest_id='$'):\n        \"\"\"Set the latest ID for a consumer group\"\"\"\n        fut = self.execute(b'XGROUP', b'SETID', stream, group_name, latest_id)\n        return wait_ok(fut)\n\n    def xgroup_destroy(self, stream, group_name):\n        \"\"\"Delete a consumer group\"\"\"\n        fut = self.execute(b'XGROUP', b'DESTROY', stream, group_name)\n        return wait_ok(fut)\n\n    def xgroup_delconsumer(self, stream, group_name, consumer_name):\n        \"\"\"Delete a specific consumer from a group\"\"\"\n        fut = self.execute(\n            b'XGROUP', b'DELCONSUMER', stream, group_name, consumer_name\n        )\n        return wait_convert(fut, int)\n\n    def xpending(self, stream, group_name, start=None, stop=None, count=None,\n                 consumer=None):\n        \"\"\"Get information on pending messages for a stream\n\n        Returned data will vary depending on the presence (or not)\n        of the start/stop/count parameters. For more details see:\n        https://redis.io/commands/xpending\n\n        :raises ValueError: if the start/stop/count parameters are only\n                            partially specified\n        \"\"\"\n        # Returns: total pel messages, min id, max id, count\n        ssc = [start, stop, count]\n        ssc_count = len([v for v in ssc if v is not None])\n        if ssc_count != 3 and ssc_count != 0:\n            raise ValueError(\n                'Either specify non or all of the start/stop/count arguments'\n            )\n        if not any(ssc):\n            ssc = []\n\n        args = [stream, group_name] + ssc\n        if consumer:\n            args.append(consumer)\n        return self.execute(b'XPENDING', *args)\n\n    def xclaim(self, stream, group_name, consumer_name, min_idle_time,\n               id, *ids):\n        \"\"\"Claim a message for a given consumer\"\"\"\n        fut = self.execute(\n            b'XCLAIM', stream, group_name, consumer_name, min_idle_time,\n            id, *ids\n        )\n        return wait_convert(fut, parse_messages)\n\n    def xack(self, stream, group_name, id, *ids):\n        \"\"\"Acknowledge a message for a given consumer group\"\"\"\n        return self.execute(b'XACK', stream, group_name, id, *ids)\n\n    def xinfo(self, stream):\n        \"\"\"Retrieve information about the given stream.\n\n        An alias for ``xinfo_stream()``\n        \"\"\"\n        return self.xinfo_stream(stream)\n\n    def xinfo_consumers(self, stream, group_name):\n        \"\"\"Retrieve consumers of a consumer group\"\"\"\n        fut = self.execute(b'XINFO', b'CONSUMERS', stream, group_name)\n\n        return wait_convert(fut, parse_lists_to_dicts)\n\n    def xinfo_groups(self, stream):\n        \"\"\"Retrieve the consumer groups for a stream\"\"\"\n        fut = self.execute(b'XINFO', b'GROUPS', stream)\n        return wait_convert(fut, parse_lists_to_dicts)\n\n    def xinfo_stream(self, stream):\n        \"\"\"Retrieve information about the given stream.\"\"\"\n        fut = self.execute(b'XINFO', b'STREAM', stream)\n        return wait_make_dict(fut)\n\n    def xinfo_help(self):\n        \"\"\"Retrieve help regarding the ``XINFO`` sub-commands\"\"\"\n        fut = self.execute(b'XINFO', b'HELP')\n        return wait_convert(fut, lambda l: b'\\n'.join(l))\n\n    def _xread(self, streams, timeout=0, count=None, latest_ids=None):\n        \"\"\"Wraps up common functionality between ``xread()``\n        and ``xread_group()``\n\n        You should probably be using ``xread()`` or ``xread_group()`` directly.\n        \"\"\"\n        if latest_ids is None:\n            latest_ids = ['$'] * len(streams)\n        if len(streams) != len(latest_ids):\n            raise ValueError(\n                'The streams and latest_ids parameters must be of the '\n                'same length'\n            )\n\n        count_args = [b'COUNT', count] if count else []\n        if timeout is None:\n            block_args = []\n        else:\n            block_args = [b'BLOCK', timeout]\n        return block_args + count_args + [b'STREAMS'] + streams + latest_ids\n", "path": "aioredis/commands/streams.py"}], "after_files": [{"content": "from collections import OrderedDict\n\nfrom aioredis.util import wait_convert, wait_make_dict, wait_ok\n\n\ndef fields_to_dict(fields, type_=OrderedDict):\n    \"\"\"Convert a flat list of key/values into an OrderedDict\"\"\"\n    fields_iterator = iter(fields)\n    return type_(zip(fields_iterator, fields_iterator))\n\n\ndef parse_messages(messages):\n    \"\"\" Parse messages as returned by Redis into something useful\n\n    Messages returned by XRANGE arrive in the form:\n\n        [\n            [message_id, [key1, value1, key2, value2, ...]],\n            ...\n        ]\n\n    Here we parse this into:\n\n        [\n            [message_id, OrderedDict(\n                (key1, value1),\n                (key2, value2),\n                ...\n            )],\n            ...\n        ]\n\n    \"\"\"\n    if messages is None:\n        return []\n    return [(mid, fields_to_dict(values)) for mid, values in messages]\n\n\ndef parse_messages_by_stream(messages_by_stream):\n    \"\"\" Parse messages returned by stream\n\n    Messages returned by XREAD arrive in the form:\n        [stream_name,\n            [\n                [message_id, [key1, value1, key2, value2, ...]],\n                ...\n            ],\n            ...\n        ]\n\n    Here we parse this into (with the help of the above parse_messages()\n    function):\n\n        [\n            [stream_name, message_id, OrderedDict(\n                (key1, value1),\n                (key2, value2),.\n                ...\n            )],\n            ...\n        ]\n\n    \"\"\"\n    if messages_by_stream is None:\n        return []\n\n    parsed = []\n    for stream, messages in messages_by_stream:\n        for message_id, fields in parse_messages(messages):\n            parsed.append((stream, message_id, fields))\n    return parsed\n\n\ndef parse_lists_to_dicts(lists):\n    \"\"\" Convert [[a, 1, b, 2], ...] into [{a:1, b: 2}, ...]\"\"\"\n    return [fields_to_dict(l, type_=dict) for l in lists]\n\n\nclass StreamCommandsMixin:\n    \"\"\"Stream commands mixin\n\n    Streams are available in Redis since v5.0\n    \"\"\"\n\n    def xadd(self, stream, fields, message_id=b'*', max_len=None,\n             exact_len=False):\n        \"\"\"Add a message to a stream.\"\"\"\n        args = []\n        if max_len is not None:\n            if exact_len:\n                args.extend((b'MAXLEN', max_len))\n            else:\n                args.extend((b'MAXLEN', b'~', max_len))\n\n        args.append(message_id)\n\n        for k, v in fields.items():\n            args.extend([k, v])\n        return self.execute(b'XADD', stream, *args)\n\n    def xrange(self, stream, start='-', stop='+', count=None):\n        \"\"\"Retrieve messages from a stream.\"\"\"\n        if count is not None:\n            extra = ['COUNT', count]\n        else:\n            extra = []\n        fut = self.execute(b'XRANGE', stream, start, stop, *extra)\n        return wait_convert(fut, parse_messages)\n\n    def xrevrange(self, stream, start='+', stop='-', count=None):\n        \"\"\"Retrieve messages from a stream in reverse order.\"\"\"\n        if count is not None:\n            extra = ['COUNT', count]\n        else:\n            extra = []\n        fut = self.execute(b'XREVRANGE', stream, start, stop, *extra)\n        return wait_convert(fut, parse_messages)\n\n    def xread(self, streams, timeout=0, count=None, latest_ids=None):\n        \"\"\"Perform a blocking read on the given stream\n\n        :raises ValueError: if the length of streams and latest_ids do\n                            not match\n        \"\"\"\n        args = self._xread(streams, timeout, count, latest_ids)\n        fut = self.execute(b'XREAD', *args)\n        return wait_convert(fut, parse_messages_by_stream)\n\n    def xread_group(self, group_name, consumer_name, streams, timeout=0,\n                    count=None, latest_ids=None):\n        \"\"\"Perform a blocking read on the given stream as part of a consumer group\n\n        :raises ValueError: if the length of streams and latest_ids do\n                            not match\n        \"\"\"\n        args = self._xread(streams, timeout, count, latest_ids)\n        fut = self.execute(\n            b'XREADGROUP', b'GROUP', group_name, consumer_name, *args\n        )\n        return wait_convert(fut, parse_messages_by_stream)\n\n    def xgroup_create(self, stream, group_name, latest_id='$', mkstream=False):\n        \"\"\"Create a consumer group\"\"\"\n        args = [b'CREATE', stream, group_name, latest_id]\n        if mkstream:\n            args.append(b'MKSTREAM')\n        fut = self.execute(b'XGROUP', *args)\n        return wait_ok(fut)\n\n    def xgroup_setid(self, stream, group_name, latest_id='$'):\n        \"\"\"Set the latest ID for a consumer group\"\"\"\n        fut = self.execute(b'XGROUP', b'SETID', stream, group_name, latest_id)\n        return wait_ok(fut)\n\n    def xgroup_destroy(self, stream, group_name):\n        \"\"\"Delete a consumer group\"\"\"\n        fut = self.execute(b'XGROUP', b'DESTROY', stream, group_name)\n        return wait_ok(fut)\n\n    def xgroup_delconsumer(self, stream, group_name, consumer_name):\n        \"\"\"Delete a specific consumer from a group\"\"\"\n        fut = self.execute(\n            b'XGROUP', b'DELCONSUMER', stream, group_name, consumer_name\n        )\n        return wait_convert(fut, int)\n\n    def xpending(self, stream, group_name, start=None, stop=None, count=None,\n                 consumer=None):\n        \"\"\"Get information on pending messages for a stream\n\n        Returned data will vary depending on the presence (or not)\n        of the start/stop/count parameters. For more details see:\n        https://redis.io/commands/xpending\n\n        :raises ValueError: if the start/stop/count parameters are only\n                            partially specified\n        \"\"\"\n        # Returns: total pel messages, min id, max id, count\n        ssc = [start, stop, count]\n        ssc_count = len([v for v in ssc if v is not None])\n        if ssc_count != 3 and ssc_count != 0:\n            raise ValueError(\n                'Either specify non or all of the start/stop/count arguments'\n            )\n        if not any(ssc):\n            ssc = []\n\n        args = [stream, group_name] + ssc\n        if consumer:\n            args.append(consumer)\n        return self.execute(b'XPENDING', *args)\n\n    def xclaim(self, stream, group_name, consumer_name, min_idle_time,\n               id, *ids):\n        \"\"\"Claim a message for a given consumer\"\"\"\n        fut = self.execute(\n            b'XCLAIM', stream, group_name, consumer_name, min_idle_time,\n            id, *ids\n        )\n        return wait_convert(fut, parse_messages)\n\n    def xack(self, stream, group_name, id, *ids):\n        \"\"\"Acknowledge a message for a given consumer group\"\"\"\n        return self.execute(b'XACK', stream, group_name, id, *ids)\n\n    def xinfo(self, stream):\n        \"\"\"Retrieve information about the given stream.\n\n        An alias for ``xinfo_stream()``\n        \"\"\"\n        return self.xinfo_stream(stream)\n\n    def xinfo_consumers(self, stream, group_name):\n        \"\"\"Retrieve consumers of a consumer group\"\"\"\n        fut = self.execute(b'XINFO', b'CONSUMERS', stream, group_name)\n\n        return wait_convert(fut, parse_lists_to_dicts)\n\n    def xinfo_groups(self, stream):\n        \"\"\"Retrieve the consumer groups for a stream\"\"\"\n        fut = self.execute(b'XINFO', b'GROUPS', stream)\n        return wait_convert(fut, parse_lists_to_dicts)\n\n    def xinfo_stream(self, stream):\n        \"\"\"Retrieve information about the given stream.\"\"\"\n        fut = self.execute(b'XINFO', b'STREAM', stream)\n        return wait_make_dict(fut)\n\n    def xinfo_help(self):\n        \"\"\"Retrieve help regarding the ``XINFO`` sub-commands\"\"\"\n        fut = self.execute(b'XINFO', b'HELP')\n        return wait_convert(fut, lambda l: b'\\n'.join(l))\n\n    def _xread(self, streams, timeout=0, count=None, latest_ids=None):\n        \"\"\"Wraps up common functionality between ``xread()``\n        and ``xread_group()``\n\n        You should probably be using ``xread()`` or ``xread_group()`` directly.\n        \"\"\"\n        if latest_ids is None:\n            latest_ids = ['$'] * len(streams)\n        if len(streams) != len(latest_ids):\n            raise ValueError(\n                'The streams and latest_ids parameters must be of the '\n                'same length'\n            )\n\n        count_args = [b'COUNT', count] if count else []\n        if timeout is None:\n            block_args = []\n        else:\n            block_args = [b'BLOCK', timeout]\n        return block_args + count_args + [b'STREAMS'] + streams + latest_ids\n", "path": "aioredis/commands/streams.py"}]}