async def update_datatype(self, datatype, **options):

        if datatype.bucket.bucket_type.is_default():
            raise NotImplementedError("Datatypes cannot be used in the default"
                                      " bucket-type.")

        op = datatype.to_op()
        type_name = datatype.type_name
        if not op:
            raise ValueError("No operation to send on datatype {!r}".
                             format(datatype))

        req = riak_pb.DtUpdateReq()
        req.bucket = str_to_bytes(datatype.bucket.name)
        req.type = str_to_bytes(datatype.bucket.bucket_type.name)

        if datatype.key:
            req.key = str_to_bytes(datatype.key)
        if datatype._context:
            req.context = datatype._context

        self._encode_dt_options(req, options)
        self._encode_dt_op(type_name, req, op)

        msg_code, resp = await self._request(
            messages.MSG_CODE_DT_UPDATE_REQ, req,
            messages.MSG_CODE_DT_UPDATE_RESP)
        if resp.HasField('key'):
            datatype.key = resp.key[:]
        if resp.HasField('context'):
            datatype._context = resp.context[:]

        datatype._set_value(self._decode_dt_value(type_name, resp))

        return True

    def delete(self, robj, rw=None, r=None, w=None, dw=None, pr=None, pw=None,
               timeout=None):
        req = riak_pb.RpbDelReq()
        if rw:
            req.rw = self._encode_quorum(rw)
        if r:
            req.r = self._encode_quorum(r)
        if w:
            req.w = self._encode_quorum(w)
        if dw:
            req.dw = self._encode_quorum(dw)

        if self.quorum_controls():
            if pr:
                req.pr = self._encode_quorum(pr)
            if pw:
                req.pw = self._encode_quorum(pw)

        if self.client_timeouts() and timeout:
            req.timeout = timeout

        use_vclocks = (self.tombstone_vclocks() and
                       hasattr(robj, 'vclock') and robj.vclock)
        if use_vclocks:
            req.vclock = robj.vclock.encode('binary')

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

        msg_code, resp = self._request(MSG_CODE_DEL_REQ, req,
                                       MSG_CODE_DEL_RESP)
        return self

    def _encode_bucket_props(self, props, msg):
        """
        Encodes a dict of bucket properties into the protobuf message.

        :param props: bucket properties
        :type props: dict
        :param msg: the protobuf message to fill
        :type msg: riak.pb.riak_pb2.RpbSetBucketReq
        """
        for prop in NORMAL_PROPS:
            if prop in props and props[prop] is not None:
                if isinstance(props[prop], string_types):
                    setattr(msg.props, prop, str_to_bytes(props[prop]))
                else:
                    setattr(msg.props, prop, props[prop])
        for prop in COMMIT_HOOK_PROPS:
            if prop in props:
                setattr(msg.props, 'has_' + prop, True)
                self._encode_hooklist(props[prop], getattr(msg.props, prop))
        for prop in MODFUN_PROPS:
            if prop in props and props[prop] is not None:
                self._encode_modfun(props[prop], getattr(msg.props, prop))
        for prop in QUORUM_PROPS:
            if prop in props and props[prop] not in (None, 'default'):
                value = self._encode_quorum(props[prop])
                if value is not None:
                    if isinstance(value, string_types):
                        setattr(msg.props, prop, str_to_bytes(value))
                    else:
                        setattr(msg.props, prop, value)
        if 'repl' in props:
            msg.props.repl = REPL_TO_PB[props['repl']]

        return msg

    def update_counter(self, bucket, key, value, **params):
        if not bucket.bucket_type.is_default():
            raise NotImplementedError("Counters are not "
                                      "supported with bucket-types, "
                                      "use datatypes instead.")

        if not self.counters():
            raise NotImplementedError("Counters are not supported")

        req = riak_pb.RpbCounterUpdateReq()
        req.bucket = str_to_bytes(bucket.name)
        req.key = str_to_bytes(key)
        req.amount = value
        if params.get('w') is not None:
            req.w = self._encode_quorum(params['w'])
        if params.get('dw') is not None:
            req.dw = self._encode_quorum(params['dw'])
        if params.get('pw') is not None:
            req.pw = self._encode_quorum(params['pw'])
        if params.get('returnvalue') is not None:
            req.returnvalue = params['returnvalue']

        msg_code, resp = self._request(MSG_CODE_COUNTER_UPDATE_REQ, req,
                                       MSG_CODE_COUNTER_UPDATE_RESP)
        if resp.HasField('value'):
            return resp.value
        else:
            return True

    def get_counter(self, bucket, key, **params):
        if not bucket.bucket_type.is_default():
            raise NotImplementedError("Counters are not " "supported with bucket-types, " "use datatypes instead.")

        if not self.counters():
            raise NotImplementedError("Counters are not supported")

        req = riak.pb.riak_kv_pb2.RpbCounterGetReq()
        req.bucket = str_to_bytes(bucket.name)
        req.key = str_to_bytes(key)
        if params.get("r") is not None:
            req.r = self._encode_quorum(params["r"])
        if params.get("pr") is not None:
            req.pr = self._encode_quorum(params["pr"])
        if params.get("basic_quorum") is not None:
            req.basic_quorum = params["basic_quorum"]
        if params.get("notfound_ok") is not None:
            req.notfound_ok = params["notfound_ok"]

        msg_code, resp = self._request(
            riak.pb.messages.MSG_CODE_COUNTER_GET_REQ, req, riak.pb.messages.MSG_CODE_COUNTER_GET_RESP
        )
        if resp.HasField("value"):
            return resp.value
        else:
            return None

 def encode_stream_mapred(self, content):
     req = riak.pb.riak_kv_pb2.RpbMapRedReq()
     req.request = str_to_bytes(content)
     req.content_type = str_to_bytes("application/json")
     mc = riak.pb.messages.MSG_CODE_MAP_RED_REQ
     rc = riak.pb.messages.MSG_CODE_MAP_RED_RESP
     return Msg(mc, req.SerializeToString(), rc)

    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 encode_delete(self, robj, rw=None, r=None,
                      w=None, dw=None, pr=None, pw=None,
                      timeout=None):
        req = riak.pb.riak_kv_pb2.RpbDelReq()
        if rw:
            req.rw = self.encode_quorum(rw)
        if r:
            req.r = self.encode_quorum(r)
        if w:
            req.w = self.encode_quorum(w)
        if dw:
            req.dw = self.encode_quorum(dw)

        if self._quorum_controls:
            if pr:
                req.pr = self.encode_quorum(pr)
            if pw:
                req.pw = self.encode_quorum(pw)

        if self._client_timeouts and timeout:
            req.timeout = timeout

        use_vclocks = (self._tombstone_vclocks and
                       hasattr(robj, 'vclock') and robj.vclock)
        if use_vclocks:
            req.vclock = robj.vclock.encode('binary')

        bucket = robj.bucket
        req.bucket = str_to_bytes(bucket.name)
        self._add_bucket_type(req, bucket.bucket_type)
        req.key = str_to_bytes(robj.key)
        mc = riak.pb.messages.MSG_CODE_DEL_REQ
        rc = riak.pb.messages.MSG_CODE_DEL_RESP
        return Msg(mc, req.SerializeToString(), rc)

 def encode_put(self, robj, w=None, dw=None, pw=None,
                return_body=True, if_none_match=False,
                timeout=None):
     bucket = robj.bucket
     req = riak.pb.riak_kv_pb2.RpbPutReq()
     if w:
         req.w = self.encode_quorum(w)
     if dw:
         req.dw = self.encode_quorum(dw)
     if self._quorum_controls and pw:
         req.pw = self.encode_quorum(pw)
     if return_body:
         req.return_body = 1
     if if_none_match:
         req.if_none_match = 1
     if self._client_timeouts and timeout:
         req.timeout = timeout
     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)
     mc = riak.pb.messages.MSG_CODE_PUT_REQ
     rc = riak.pb.messages.MSG_CODE_PUT_RESP
     return Msg(mc, req.SerializeToString(), rc)

    def get_counter(self, bucket, key, **params):
        if not bucket.bucket_type.is_default():
            raise NotImplementedError("Counters are not "
                                      "supported with bucket-types, "
                                      "use datatypes instead.")

        if not self.counters():
            raise NotImplementedError("Counters are not supported")

        req = riak_pb.RpbCounterGetReq()
        req.bucket = str_to_bytes(bucket.name)
        req.key = str_to_bytes(key)
        if params.get('r') is not None:
            req.r = self._encode_quorum(params['r'])
        if params.get('pr') is not None:
            req.pr = self._encode_quorum(params['pr'])
        if params.get('basic_quorum') is not None:
            req.basic_quorum = params['basic_quorum']
        if params.get('notfound_ok') is not None:
            req.notfound_ok = params['notfound_ok']

        msg_code, resp = self._request(MSG_CODE_COUNTER_GET_REQ, req,
                                       MSG_CODE_COUNTER_GET_RESP)
        if resp.HasField('value'):
            return resp.value
        else:
            return None

 def encode_auth(self, username, password):
     req = riak.pb.riak_pb2.RpbAuthReq()
     req.user = str_to_bytes(username)
     req.password = str_to_bytes(password)
     mc = riak.pb.messages.MSG_CODE_AUTH_REQ
     rc = riak.pb.messages.MSG_CODE_AUTH_RESP
     return Msg(mc, req.SerializeToString(), rc)

 def encode_get_preflist(self, bucket, key):
     req = riak.pb.riak_kv_pb2.RpbGetBucketKeyPreflistReq()
     req.bucket = str_to_bytes(bucket.name)
     req.key = str_to_bytes(key)
     req.type = str_to_bytes(bucket.bucket_type.name)
     mc = riak.pb.messages.MSG_CODE_GET_BUCKET_KEY_PREFLIST_REQ
     rc = riak.pb.messages.MSG_CODE_GET_BUCKET_KEY_PREFLIST_RESP
     return Msg(mc, req.SerializeToString(), rc)

 def encode_search(self, index, query, **kwargs):
     req = riak.pb.riak_search_pb2.RpbSearchQueryReq(
             index=str_to_bytes(index),
             q=str_to_bytes(query))
     self.encode_search_query(req, **kwargs)
     mc = riak.pb.messages.MSG_CODE_SEARCH_QUERY_REQ
     rc = riak.pb.messages.MSG_CODE_SEARCH_QUERY_RESP
     return Msg(mc, req.SerializeToString(), rc)

 def encode_create_search_schema(self, schema, content):
     scma = riak.pb.riak_yokozuna_pb2.RpbYokozunaSchema(
             name=str_to_bytes(schema),
             content=str_to_bytes(content))
     req = riak.pb.riak_yokozuna_pb2.RpbYokozunaSchemaPutReq(
             schema=scma)
     mc = riak.pb.messages.MSG_CODE_YOKOZUNA_SCHEMA_PUT_REQ
     rc = riak.pb.messages.MSG_CODE_PUT_RESP
     return Msg(mc, req.SerializeToString(), rc)

 def encode_fetch_datatype(self, bucket, key, **kwargs):
     req = riak.pb.riak_dt_pb2.DtFetchReq()
     req.type = str_to_bytes(bucket.bucket_type.name)
     req.bucket = str_to_bytes(bucket.name)
     req.key = str_to_bytes(key)
     self.encode_dt_options(req, **kwargs)
     mc = riak.pb.messages.MSG_CODE_DT_FETCH_REQ
     rc = riak.pb.messages.MSG_CODE_DT_FETCH_RESP
     return Msg(mc, req.SerializeToString(), rc)

    def test_encode_data_for_get(self):
        keylist = [
            str_to_bytes('hash1'), str_to_bytes('user2'), unix_time_millis(ts0)
        ]
        req = tsgetreq_a, str_to_bytes(table_name), keylist, udef_a
        req_test = encode(req)

        test_key = ['hash1', 'user2', ts0]
        c = TtbCodec()
        msg = c.encode_timeseries_keyreq(self.table, test_key)
        self.assertEqual(req_test, msg.data)

    def create_search_schema(self, schema, content):
        if not self.pb_search_admin():
            raise NotImplementedError("Search 2.0 administration is not "
                                      "supported for this version")
        scma = riak_pb.RpbYokozunaSchema(name=str_to_bytes(schema),
                                         content=str_to_bytes(content))
        req = riak_pb.RpbYokozunaSchemaPutReq(schema=scma)

        self._request(MSG_CODE_YOKOZUNA_SCHEMA_PUT_REQ, req,
                      MSG_CODE_PUT_RESP)
        return True

    def stream_mapred(self, inputs, query, timeout=None):
        # Construct the job, optionally set the timeout...
        content = self._construct_mapred_json(inputs, query, timeout)

        req = riak_pb.RpbMapRedReq()
        req.request = str_to_bytes(content)
        req.content_type = str_to_bytes("application/json")

        self._send_msg(MSG_CODE_MAP_RED_REQ, req)

        return RiakPbcMapredStream(self)

 def encode_create_search_index(self, index, schema=None,
                                n_val=None, timeout=None):
     index = str_to_bytes(index)
     idx = riak.pb.riak_yokozuna_pb2.RpbYokozunaIndex(name=index)
     if schema:
         idx.schema = str_to_bytes(schema)
     if n_val:
         idx.n_val = n_val
     req = riak.pb.riak_yokozuna_pb2.RpbYokozunaIndexPutReq(index=idx)
     if timeout is not None:
         req.timeout = timeout
     mc = riak.pb.messages.MSG_CODE_YOKOZUNA_INDEX_PUT_REQ
     rc = riak.pb.messages.MSG_CODE_PUT_RESP
     return Msg(mc, req.SerializeToString(), rc)

    def create_search_index(self, index, schema=None, n_val=None):
        if not self.pb_search_admin():
            raise NotImplementedError("Search 2.0 administration is not "
                                      "supported for this version")
        index = str_to_bytes(index)
        idx = riak_pb.RpbYokozunaIndex(name=index)
        if schema:
            idx.schema = str_to_bytes(schema)
        if n_val:
            idx.n_val = n_val
        req = riak_pb.RpbYokozunaIndexPutReq(index=idx)

        self._request(MSG_CODE_YOKOZUNA_INDEX_PUT_REQ, req,
                      MSG_CODE_PUT_RESP)
        return True