def xreadgroup(self, groupname, consumername, streams, count=None,
                   block=None, noack=False):
        """
        Read from a stream via a consumer group.
        groupname: name of the consumer group.
        consumername: name of the requesting consumer.
        streams: a dict of stream names to stream IDs, where
               IDs indicate the last ID already seen.
        count: if set, only return this many items, beginning with the
               earliest available.
        block: number of milliseconds to wait, if nothing already present.
        noack: do not add messages to the PEL
        """
        pieces = [Token.get_token('GROUP'), groupname, consumername]
        if count is not None:
            if not isinstance(count, (int, long)) or count < 1:
                raise DataError("XREADGROUP count must be a positive integer")
            pieces.append(Token.get_token("COUNT"))
            pieces.append(str(count))
        if block is not None:
            if not isinstance(block, (int, long)) or block < 0:
                raise DataError("XREADGROUP block must be a non-negative "
                                "integer")
            pieces.append(Token.get_token("BLOCK"))
            pieces.append(str(block))
        if noack:
            pieces.append(Token.get_token("NOACK"))
        if not isinstance(streams, dict) or len(streams) == 0:
            raise DataError('XREADGROUP streams must be a non empty dict')
        pieces.append(Token.get_token('STREAMS'))
        pieces.extend(streams.keys())
        pieces.extend(streams.values())
        return self.execute_command('XREADGROUP', *pieces)