def get_index(self, bucket, index, startkey, endkey=None,
                  return_terms=None, max_results=None, continuation=None,
                  timeout=None, term_regex=None):
        if not self.pb_indexes():
            return self._get_index_mapred_emu(bucket, index, startkey, endkey)

        if term_regex and not self.index_term_regex():
            raise NotImplementedError("Secondary index term_regex is not "
                                      "supported")

        req = self._encode_index_req(bucket, index, startkey, endkey,
                                     return_terms, max_results, continuation,
                                     timeout, term_regex)

        msg_code, resp = self._request(MSG_CODE_INDEX_REQ, req,
                                       MSG_CODE_INDEX_RESP)

        if return_terms and resp.results:
            results = [(decode_index_value(index, pair.key),
                        bytes_to_str(pair.value))
                       for pair in resp.results]
        else:
            results = resp.keys[:]
            if PY3:
                results = [bytes_to_str(key) for key in resp.keys]

        if max_results is not None and resp.HasField('continuation'):
            return (results, bytes_to_str(resp.continuation))
        else:
            return (results, None)

 def get_server_info(self):
     """
     Get information about the server
     """
     msg_code, resp = self._request(MSG_CODE_GET_SERVER_INFO_REQ,
                                    expect=MSG_CODE_GET_SERVER_INFO_RESP)
     return {'node': bytes_to_str(resp.node),
             'server_version': bytes_to_str(resp.server_version)}

 def get_server_info(self):
     """
     Get information about the server
     """
     msg_code, resp = self._request(
         riak.pb.messages.MSG_CODE_GET_SERVER_INFO_REQ, expect=riak.pb.messages.MSG_CODE_GET_SERVER_INFO_RESP
     )
     return {"node": bytes_to_str(resp.node), "server_version": bytes_to_str(resp.server_version)}

    def _decode_modfun(self, modfun):
        """
        Decodes a protobuf modfun pair into a dict with 'mod' and
        'fun' keys. Used in bucket properties.

        :param modfun: the protobuf message to decode
        :type modfun: riak_pb.RpbModFun
        :rtype dict
        """
        return {"mod": bytes_to_str(modfun.module), "fun": bytes_to_str(modfun.function)}

 def _decode_search_doc(self, doc):
     resultdoc = MultiDict()
     for pair in doc.fields:
         if PY2:
             ukey = unicode(pair.key, 'utf-8')    # noqa
             uval = unicode(pair.value, 'utf-8')  # noqa
         else:
             ukey = bytes_to_str(pair.key)
             uval = bytes_to_str(pair.value)
         resultdoc.add(ukey, uval)
     return resultdoc.mixed()

    def get_search_schema(self, schema):
        if not self.pb_search_admin():
            raise NotImplementedError("Search 2.0 administration is not "
                                      "supported for this version")
        req = riak_pb.RpbYokozunaSchemaGetReq(name=str_to_bytes(schema))

        msg_code, resp = self._request(MSG_CODE_YOKOZUNA_SCHEMA_GET_REQ, req,
                                       MSG_CODE_YOKOZUNA_SCHEMA_GET_RESP)
        result = {}
        result['name'] = bytes_to_str(resp.schema.name)
        result['content'] = bytes_to_str(resp.schema.content)
        return result

 def test_stream_keys(self):
     table = Table(self.client, table_name)
     streamed_keys = []
     for keylist in table.stream_keys():
         self.assertNotEqual([], keylist)
         streamed_keys += keylist
         for key in keylist:
             self.assertIsInstance(key, list)
             self.assertEqual(len(key), 3)
             self.assertEqual(bytes_to_str(key[0]), 'hash1')
             self.assertEqual(bytes_to_str(key[1]), 'user2')
             self.assertIsInstance(key[2], datetime.datetime)
     self.assertGreater(len(streamed_keys), 0)

 def _parse_msg(self):
     ''' Parse protobuf message'''
     self._msg = self._data[
         self.HEADER_LENGTH:self.HEADER_LENGTH + self._msglen]
     self.msg_code, = struct.unpack("B", self._msg[:1])
     if self.msg_code is messages.MSG_CODE_ERROR_RESP:
         error = self._get_pb_msg(self.msg_code, self._msg[1:])
         logger.error('Riak error message recieved: %s',
                      bytes_to_str(error.errmsg))
         raise RiakError(bytes_to_str(error.errmsg))
     elif self.msg_code in messages.MESSAGE_CLASSES:
         logger.debug('Normal message with code %d received', self.msg_code)
         self.msg = self._get_pb_msg(self.msg_code, self._msg[1:])
     else:
         logger.error('Unknown message received [%d]', self.msg_code)

    def _decode_search_index(self, index):
        """
        Fills an RpbYokozunaIndex message with the appropriate data.

        :param index: a yz index message
        :type index: riak_pb.RpbYokozunaIndex
        :rtype dict
        """
        result = {}
        result["name"] = bytes_to_str(index.name)
        if index.HasField("schema"):
            result["schema"] = bytes_to_str(index.schema)
        if index.HasField("n_val"):
            result["n_val"] = index.n_val
        return result

    def decode_index_req(self, resp, index,
                         return_terms=None, max_results=None):
        if return_terms and resp.results:
            results = [(decode_index_value(index, pair.key),
                        bytes_to_str(pair.value))
                       for pair in resp.results]
        else:
            results = resp.keys[:]
            if six.PY3:
                results = [bytes_to_str(key) for key in resp.keys]

        if max_results is not None and resp.HasField('continuation'):
            return (results, bytes_to_str(resp.continuation))
        else:
            return (results, None)

    def _decode_search_index(self, index):
        """
        Fills an RpbYokozunaIndex message with the appropriate data.

        :param index: a yz index message
        :type index: riak.pb.riak_yokozuna_pb2.RpbYokozunaIndex
        :rtype dict
        """
        result = {}
        result['name'] = bytes_to_str(index.name)
        if index.HasField('schema'):
            result['schema'] = bytes_to_str(index.schema)
        if index.HasField('n_val'):
            result['n_val'] = index.n_val
        return result

    def decode_timeseries(self, resp, tsobj,
                          convert_timestamp=False):
        """
        Fills an TsObject with the appropriate data and
        metadata from a TsGetResp / TsQueryResp.

        :param resp: the protobuf message from which to process data
        :type resp: riak.pb.riak_ts_pb2.TsQueryRsp or
                    riak.pb.riak_ts_pb2.TsGetResp
        :param tsobj: a TsObject
        :type tsobj: TsObject
        :param convert_timestamp: Convert timestamps to datetime objects
        :type tsobj: boolean
        """
        if resp.columns is not None:
            col_names = []
            col_types = []
            for col in resp.columns:
                col_names.append(bytes_to_str(col.name))
                col_type = self.decode_timeseries_col_type(col.type)
                col_types.append(col_type)
            tsobj.columns = TsColumns(col_names, col_types)

        tsobj.rows = []
        if resp.rows is not None:
            for row in resp.rows:
                tsobj.rows.append(
                    self.decode_timeseries_row(
                        row, resp.columns, convert_timestamp))

    def get_buckets(self, transport, bucket_type=None, timeout=None):
        """
        get_buckets(bucket_type=None, timeout=None)

        Get the list of buckets as :class:`RiakBucket
        <riak.bucket.RiakBucket>` instances.

        .. warning:: Do not use this in production, as it requires
           traversing through all keys stored in a cluster.

        .. note:: This request is automatically retried :attr:`retries`
           times if it fails due to network error.

        :param bucket_type: the optional containing bucket type
        :type bucket_type: :class:`~riak.bucket.BucketType`
        :param timeout: a timeout value in milliseconds
        :type timeout: int
        :rtype: list of :class:`RiakBucket <riak.bucket.RiakBucket>`
                instances
        """
        _validate_timeout(timeout)
        if bucket_type:
            bucketfn = self._bucket_type_bucket_builder
        else:
            bucketfn = self._default_type_bucket_builder

        return [bucketfn(bytes_to_str(name), bucket_type) for name in
                transport.get_buckets(bucket_type=bucket_type,
                                      timeout=timeout)]

    def get_index(self, bucket, index, startkey, endkey=None,
                  return_terms=None, max_results=None, continuation=None,
                  timeout=None, term_regex=None):
        """
        Performs a secondary index query.
        """
        if term_regex and not self.index_term_regex():
            raise NotImplementedError("Secondary index term_regex is not "
                                      "supported on %s" %
                                      self.server_version.vstring)

        if timeout == 'infinity':
            timeout = 0

        params = {'return_terms': return_terms, 'max_results': max_results,
                  'continuation': continuation, 'timeout': timeout,
                  'term_regex': term_regex}
        bucket_type = self._get_bucket_type(bucket.bucket_type)
        url = self.index_path(bucket.name, index, startkey, endkey,
                              bucket_type=bucket_type, **params)
        status, headers, body = self._request('GET', url)
        self.check_http_code(status, [200])
        json_data = json.loads(bytes_to_str(body))
        if return_terms and u'results' in json_data:
            results = []
            for result in json_data[u'results'][:]:
                term, key = list(result.items())[0]
                results.append((decode_index_value(index, term), key),)
        else:
            results = json_data[u'keys'][:]

        if max_results and u'continuation' in json_data:
            return (results, json_data[u'continuation'])
        else:
            return (results, None)

    def next(self):
        response = super(RiakPbcMapredStream, self).next()

        if response.done and not response.HasField('response'):
            raise StopIteration

        return response.phase, json.loads(bytes_to_str(response.response))

    def _decode_bucket_props(self, msg):
        """
        Decodes the protobuf bucket properties message into a dict.

        :param msg: the protobuf message to decode
        :type msg: riak.pb.riak_pb2.RpbBucketProps
        :rtype dict
        """
        props = {}

        for prop in NORMAL_PROPS:
            if msg.HasField(prop):
                props[prop] = getattr(msg, prop)
                if isinstance(props[prop], bytes):
                    props[prop] = bytes_to_str(props[prop])
        for prop in COMMIT_HOOK_PROPS:
            if getattr(msg, 'has_' + prop):
                props[prop] = self._decode_hooklist(getattr(msg, prop))
        for prop in MODFUN_PROPS:
            if msg.HasField(prop):
                props[prop] = self._decode_modfun(getattr(msg, prop))
        for prop in QUORUM_PROPS:
            if msg.HasField(prop):
                props[prop] = self._decode_quorum(getattr(msg, prop))
        if msg.HasField('repl'):
            props['repl'] = REPL_TO_PY[msg.repl]

        return props

def mkpath(*segments, **query):
    """
    Constructs the path & query portion of a URI from path segments
    and a dict.
    """
    # Remove empty segments (e.g. no key specified)
    segments = [bytes_to_str(s) for s in segments if s is not None]
    # Join the segments into a path
    pathstring = '/'.join(segments)
    # Remove extra slashes
    pathstring = re.sub('/+', '/', pathstring)

    # Add the query string if it exists
    _query = {}
    for key in query:
        if query[key] in [False, True]:
            _query[key] = str(query[key]).lower()
        elif query[key] is not None:
            if PY2 and isinstance(query[key], unicode):  # noqa
                _query[key] = query[key].encode('utf-8')
            else:
                _query[key] = query[key]

    if len(_query) > 0:
        pathstring += "?" + urlencode(_query)

    if not pathstring.startswith('/'):
        pathstring = '/' + pathstring

    return pathstring

    async def put(self, robj, return_body=True):
        bucket = robj.bucket

        req = riak_pb.RpbPutReq()

        if return_body:
            req.return_body = 1

        req.bucket = str_to_bytes(bucket.name)
        self._add_bucket_type(req, bucket.bucket_type)

        if robj.key:
            req.key = str_to_bytes(robj.key)
        if robj.vclock:
            req.vclock = robj.vclock.encode('binary')

        self._encode_content(robj, req.content)

        msg_code, resp = await self._request(messages.MSG_CODE_PUT_REQ, req,
                                             messages.MSG_CODE_PUT_RESP)

        if resp is not None:
            if resp.HasField('key'):
                robj.key = bytes_to_str(resp.key)
            if resp.HasField('vclock'):
                robj.vclock = VClock(resp.vclock, 'binary')
            if resp.content:
                self._decode_contents(resp.content, robj)
        elif not robj.key:
            raise RiakError("missing response object")

        return robj

 def maybe_riak_error(self, msg_code, data=None):
     if msg_code == riak.pb.messages.MSG_CODE_ERROR_RESP:
         if data is None:
             raise RiakError('no error provided!')
         else:
             err = parse_pbuf_msg(msg_code, data)
             raise RiakError(bytes_to_str(err.errmsg))

    def _parse_body(self, robj, response, expected_statuses):
        """
        Parse the body of an object response and populate the object.
        """
        # If no response given, then return.
        if response is None:
            return None

        status, headers, data = response

        # Check if the server is down(status==0)
        if not status:
            m = 'Could not contact Riak Server: http://{0}:{1}!'.format(
                self._node.host, self._node.http_port)
            raise RiakError(m)

        # Make sure expected code came back
        self.check_http_code(status, expected_statuses)

        if 'x-riak-vclock' in headers:
            robj.vclock = VClock(headers['x-riak-vclock'], 'base64')

        # If 404(Not Found), then clear the object.
        if status == 404:
            robj.siblings = []
            return None
        # If 201 Created, we need to extract the location and set the
        # key on the object.
        elif status == 201:
            robj.key = headers['location'].strip().split('/')[-1]
        # If 300(Siblings), apply the siblings to the object
        elif status == 300:
            ctype, params = parse_header(headers['content-type'])
            if ctype == 'multipart/mixed':
                if six.PY3:
                    data = bytes_to_str(data)
                boundary = re.compile('\r?\n--%s(?:--)?\r?\n' %
                                      re.escape(params['boundary']))
                parts = [message_from_string(p)
                         for p in re.split(boundary, data)[1:-1]]
                robj.siblings = [self._parse_sibling(RiakContent(robj),
                                                     part.items(),
                                                     part.get_payload())
                                 for part in parts]

                # Invoke sibling-resolution logic
                if robj.resolver is not None:
                    robj.resolver(robj)

                return robj
            else:
                raise Exception('unexpected sibling response format: {0}'.
                                format(ctype))

        robj.siblings = [self._parse_sibling(RiakContent(robj),
                                             headers.items(),
                                             data)]

        return robj