mirror
/
etcd-cpp-apiv3
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Updated client::set() function to use grpc. 

Parse response for PutRequest is still stubbed.
This commit is contained in:
arches 2016-06-01 08:07:09 -04:00
parent 221cdf39f2
commit c75548d9e3
2 changed files with 70 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
etcd/Client.hpp
View File

@ -10,6 +10,12 @@
#include "proto/rpc.grpc.pb.h" #include "proto/rpc.grpc.pb.h"
using grpc::Channel; using grpc::Channel;
using grpc::ClientAsyncResponseReader;
using grpc::ClientContext;
using grpc::CompletionQueue;
using grpc::Status;
using etcdserverpb::PutRequest;
using etcdserverpb::PutResponse;
using etcdserverpb::KV; using etcdserverpb::KV;
namespace etcd namespace etcd
@ -140,6 +146,18 @@ namespace etcd
web::http::client::http_client client; web::http::client::http_client client;
std::unique_ptr<KV::Stub> stub_; std::unique_ptr<KV::Stub> stub_;
pplx::task<etcd::Response> send_put(const std::string& key, const std::string& value);
};
class AsyncPutResponse
{
public:
PutResponse reply;
Status status;
ClientContext context;
CompletionQueue cq_;
std::unique_ptr<ClientAsyncResponseReader<PutResponse>> response_reader;
Response ParseResponse();
}; };
} }

63
src/Client.cpp
View File

@ -3,15 +3,15 @@
etcd::Client::Client(std::string const & address) etcd::Client::Client(std::string const & address)
: client(address) : client(address)
{ {
std::string stripped_address(address); std::string stripped_address(address);
std::string substr("http://"); std::string substr("http://");
std::string::size_type i = stripped_address.find(substr); std::string::size_type i = stripped_address.find(substr);
if(i != std::string::npos) if(i != std::string::npos)
{ {
stripped_address.erase(i,substr.length()); stripped_address.erase(i,substr.length());
} }
std::shared_ptr<Channel> channel = grpc::CreateChannel(stripped_address, grpc::InsecureChannelCredentials()); std::shared_ptr<Channel> channel = grpc::CreateChannel(stripped_address, grpc::InsecureChannelCredentials());
stub_= KV::NewStub(channel); stub_= KV::NewStub(channel);
} }
pplx::task<etcd::Response> etcd::Client::send_get_request(web::http::uri_builder & uri) pplx::task<etcd::Response> etcd::Client::send_get_request(web::http::uri_builder & uri)
@ -39,8 +39,7 @@ pplx::task<etcd::Response> etcd::Client::get(std::string const & key)
pplx::task<etcd::Response> etcd::Client::set(std::string const & key, std::string const & value) pplx::task<etcd::Response> etcd::Client::set(std::string const & key, std::string const & value)
{ {
web::http::uri_builder uri("/v2/keys" + key); return send_put(key,value);
return send_put_request(uri, "value", value);
} }
pplx::task<etcd::Response> etcd::Client::add(std::string const & key, std::string const & value) pplx::task<etcd::Response> etcd::Client::add(std::string const & key, std::string const & value)
@ -134,3 +133,45 @@ pplx::task<etcd::Response> etcd::Client::watch(std::string const & key, int from
uri.append_query("recursive=true"); uri.append_query("recursive=true");
return send_get_request(uri); return send_get_request(uri);
} }
etcd::Response etcd::AsyncPutResponse::ParseResponse()
{
return etcd::Response();
}
pplx::task<etcd::Response> etcd::Client::send_put(std::string const & key, std::string const & value)
{
PutRequest request;
request.set_key(key);
request.set_value(value);
etcd::AsyncPutResponse* call= new etcd::AsyncPutResponse();
call->response_reader = stub_->AsyncPut(&call->context,request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call);
return pplx::task<etcd::Response>([call]()
{
void* got_tag;
bool ok = false;
etcd::Response resp;
//blocking
call->cq_.Next(&got_tag, &ok);
GPR_ASSERT(got_tag == (void*)call);
GPR_ASSERT(ok);
etcd::AsyncPutResponse* call = static_cast<etcd::AsyncPutResponse*>(got_tag);
if(call->status.ok())
{
resp = call->ParseResponse();
}
delete call;
return resp;
});
}