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

Refactored Client class to remove boilerplate code. 

Responsibility of boilerplate work is transfered to its own class 
(Transaction).
UT for watch are temporarily commented out as they are not supported yet
This commit is contained in:
lampayan 2016-06-21 15:33:11 +02:00
parent 01bf49bf2d
commit fe5396e437
5 changed files with 342 additions and 361 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/CMakeLists.txt
View File

@ -1,4 +1,4 @@
add_library(etcd-cpp-api SHARED ../proto/kv.pb.cc ../proto/auth.pb.cc ../proto/rpc.pb.cc ../proto/rpc.grpc.pb.cc ../v3/src/Utils.cpp ../v3/src/grpcClient.cpp ../v3/src/AsyncTxnResponse.cpp ../v3/src/AsyncRangeResponse.cpp ../v3/src/AsyncPutResponse.cpp Client.cpp Response.cpp Value.cpp json_constants.cpp) add_library(etcd-cpp-api SHARED ../proto/kv.pb.cc ../proto/auth.pb.cc ../proto/rpc.pb.cc ../proto/rpc.grpc.pb.cc ../v3/src/Utils.cpp ../v3/src/grpcClient.cpp ../v3/src/AsyncTxnResponse.cpp ../v3/src/AsyncRangeResponse.cpp ../v3/src/AsyncPutResponse.cpp ../v3/src/Transaction.cpp Client.cpp Response.cpp Value.cpp json_constants.cpp)
set_property(TARGET etcd-cpp-api PROPERTY CXX_STANDARD 11) set_property(TARGET etcd-cpp-api PROPERTY CXX_STANDARD 11)
target_link_libraries(etcd-cpp-api ${CPPREST_LIB} boost_system ssl crypto protobuf grpc++) target_link_libraries(etcd-cpp-api ${CPPREST_LIB} boost_system ssl crypto protobuf grpc++)

348
src/Client.cpp
View File

@ -3,6 +3,7 @@
#include "v3/include/AsyncTxnResponse.hpp" #include "v3/include/AsyncTxnResponse.hpp"
#include "v3/include/AsyncRangeResponse.hpp" #include "v3/include/AsyncRangeResponse.hpp"
#include "v3/include/Utils.hpp" #include "v3/include/Utils.hpp"
#include "v3/include/Transaction.hpp"
#include <iostream> #include <iostream>
using grpc::Channel; using grpc::Channel;
@ -139,84 +140,33 @@ pplx::task<etcd::Response> etcd::Client::watch(std::string const & key, int from
pplx::task<etcd::Response> etcd::Client::send_asyncadd(std::string const & key, std::string const & value) pplx::task<etcd::Response> etcd::Client::send_asyncadd(std::string const & key, std::string const & value)
{ {
etcdv3::Transaction transaction(key);
transaction.init_compare(Compare::CompareResult::Compare_CompareResult_EQUAL,
Compare::CompareTarget::Compare_CompareTarget_VERSION);
//check if key is not present transaction.setup_basic_failure_operation(key);
TxnRequest txn_request; transaction.setup_basic_create_sequence(key, value);
Compare* compare = txn_request.add_compare();
compare->set_result(Compare::CompareResult::Compare_CompareResult_EQUAL);
compare->set_target(Compare::CompareTarget::Compare_CompareTarget_VERSION);
compare->set_key(key);
compare->set_version(0);
//get key on failure
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
get_request->set_key(key);
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
//if success, add key and then get new value of key
std::unique_ptr<PutRequest> put_request(new PutRequest());
put_request->set_key(key);
put_request->set_value(value);
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_put(put_request.release());
get_request.reset(new RangeRequest());
get_request->set_key(key);
req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
//to be done in one method, where txn_request is a field of a class
std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("create")); std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("create"));
call->response_reader = stub_->AsyncTxn(&call->context,transaction.txn_request,&call->cq_);
call->response_reader = stub_->AsyncTxn(&call->context,txn_request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call.get()); call->response_reader->Finish(&call->reply, &call->status, (void*)call.get());
return Response::create(call); return Response::create(call);
} }
pplx::task<etcd::Response> etcd::Client::send_asyncmodify_if(std::string const & key, std::string const & value, std::string const & old_value) pplx::task<etcd::Response> etcd::Client::send_asyncmodify_if(std::string const & key, std::string const & value, std::string const & old_value)
{ {
//check key value is equal to old_value etcdv3::Transaction transaction(key);
TxnRequest txn_request; transaction.init_compare(old_value, Compare::CompareResult::Compare_CompareResult_EQUAL,
Compare* compare = txn_request.add_compare(); Compare::CompareTarget::Compare_CompareTarget_VALUE);
compare->set_result(Compare::CompareResult::Compare_CompareResult_EQUAL);
compare->set_target(Compare::CompareTarget::Compare_CompareTarget_VALUE);
compare->set_key(key);
compare->set_value(old_value);
//get key on failure transaction.setup_basic_failure_operation(key);
std::unique_ptr<RangeRequest> get_request(new RangeRequest()); transaction.setup_compare_and_swap_sequence(value);
get_request->set_key(key);
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
//on success get key value then modify and get new value
get_request.reset(new RangeRequest());
get_request->set_key(key);
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::unique_ptr<PutRequest> put_request(new PutRequest());
put_request->set_key(key);
put_request->set_value(value);
req_success = txn_request.add_success();
req_success->set_allocated_request_put(put_request.release());
get_request.reset(new RangeRequest());
get_request->set_key(key);
req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("compareAndSwap")); std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("compareAndSwap"));
call->response_reader = stub_->AsyncTxn(&call->context,transaction.txn_request,&call->cq_);
call->response_reader = stub_->AsyncTxn(&call->context,txn_request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call.get()); call->response_reader->Finish(&call->reply, &call->status, (void*)call.get());
return Response::create(call); return Response::create(call);
@ -224,41 +174,15 @@ pplx::task<etcd::Response> etcd::Client::send_asyncmodify_if(std::string const &
pplx::task<etcd::Response> etcd::Client::send_asyncmodify_if(std::string const & key, std::string const & value, int old_index) pplx::task<etcd::Response> etcd::Client::send_asyncmodify_if(std::string const & key, std::string const & value, int old_index)
{ {
//check key value is equal to old_value etcdv3::Transaction transaction(key);
TxnRequest txn_request; transaction.init_compare(old_index, Compare::CompareResult::Compare_CompareResult_EQUAL,
Compare* compare = txn_request.add_compare(); Compare::CompareTarget::Compare_CompareTarget_MOD);
compare->set_result(Compare::CompareResult::Compare_CompareResult_EQUAL);
compare->set_target(Compare::CompareTarget::Compare_CompareTarget_MOD);
compare->set_key(key);
compare->set_mod_revision(old_index);
//get key on failure transaction.setup_basic_failure_operation(key);
std::unique_ptr<RangeRequest> get_request(new RangeRequest()); transaction.setup_compare_and_swap_sequence(value);
get_request->set_key(key);
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
//on success get key value then modify and get new value
get_request.reset(new RangeRequest());
get_request->set_key(key);
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::unique_ptr<PutRequest> put_request(new PutRequest());
put_request->set_key(key);
put_request->set_value(value);
req_success = txn_request.add_success();
req_success->set_allocated_request_put(put_request.release());
get_request.reset(new RangeRequest());
get_request->set_key(key);
req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("compareAndSwap")); std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("compareAndSwap"));
call->response_reader = stub_->AsyncTxn(&call->context,transaction.txn_request,&call->cq_);
call->response_reader = stub_->AsyncTxn(&call->context,txn_request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call.get()); call->response_reader->Finish(&call->reply, &call->status, (void*)call.get());
return Response::create(call); return Response::create(call);
@ -267,44 +191,15 @@ pplx::task<etcd::Response> etcd::Client::send_asyncmodify_if(std::string const &
pplx::task<etcd::Response> etcd::Client::send_asyncmodify(std::string const & key, std::string const & value) pplx::task<etcd::Response> etcd::Client::send_asyncmodify(std::string const & key, std::string const & value)
{ {
//check if key is present etcdv3::Transaction transaction(key);
TxnRequest txn_request; transaction.init_compare(Compare::CompareResult::Compare_CompareResult_GREATER,
Compare* compare = txn_request.add_compare(); Compare::CompareTarget::Compare_CompareTarget_VERSION);
compare->set_result(Compare::CompareResult::Compare_CompareResult_GREATER);
compare->set_target(Compare::CompareTarget::Compare_CompareTarget_VERSION);
compare->set_key(key);
compare->set_version(0);
//success or failure
//get key value before modification
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
get_request->set_key(key);
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
get_request.reset(new RangeRequest());
get_request->set_key(key);
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
//if success, modify key and then get new value of key
std::unique_ptr<PutRequest> put_request(new PutRequest());
put_request->set_key(key);
put_request->set_value(value);
req_success = txn_request.add_success();
req_success->set_allocated_request_put(put_request.release());
get_request.reset(new RangeRequest());
get_request->set_key(key);
req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
transaction.setup_basic_failure_operation(key);
transaction.setup_compare_and_swap_sequence(value);
std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("update")); std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("update"));
call->response_reader = stub_->AsyncTxn(&call->context,transaction.txn_request,&call->cq_);
call->response_reader = stub_->AsyncTxn(&call->context,txn_request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call.get()); call->response_reader->Finish(&call->reply, &call->status, (void*)call.get());
return Response::create(call); return Response::create(call);
@ -323,9 +218,7 @@ pplx::task<etcd::Response> etcd::Client::send_asyncget(std::string const & key,
} }
std::shared_ptr<etcdv3::AsyncRangeResponse> call(new etcdv3::AsyncRangeResponse()); std::shared_ptr<etcdv3::AsyncRangeResponse> call(new etcdv3::AsyncRangeResponse());
call->response_reader = stub_->AsyncRange(&call->context,get_request,&call->cq_); call->response_reader = stub_->AsyncRange(&call->context,get_request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call.get()); call->response_reader->Finish(&call->reply, &call->status, (void*)call.get());
return Response::create(call); return Response::create(call);
@ -334,48 +227,15 @@ pplx::task<etcd::Response> etcd::Client::send_asyncget(std::string const & key,
pplx::task<etcd::Response> etcd::Client::send_asyncput(std::string const & key, std::string const & value) pplx::task<etcd::Response> etcd::Client::send_asyncput(std::string const & key, std::string const & value)
{ {
//check if key is not present etcdv3::Transaction transaction(key);
TxnRequest txn_request; transaction.init_compare(Compare::CompareResult::Compare_CompareResult_EQUAL,
Compare* compare = txn_request.add_compare(); Compare::CompareTarget::Compare_CompareTarget_VERSION);
compare->set_result(Compare::CompareResult::Compare_CompareResult_EQUAL);
compare->set_target(Compare::CompareTarget::Compare_CompareTarget_VERSION);
compare->set_key(key);
compare->set_version(0);
transaction.setup_set_failure_operation(key, value);
//get key on failure, get key before put, modify and then get updated key transaction.setup_basic_create_sequence(key, value);
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
get_request->set_key(key);
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
std::unique_ptr<PutRequest> put_request(new PutRequest());
put_request->set_key(key);
put_request->set_value(value);
req_failure = txn_request.add_failure();
req_failure->set_allocated_request_put(put_request.release());
get_request.reset(new RangeRequest());
get_request->set_key(key);
req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
//if success, put key and then get new value of key
put_request.reset(new PutRequest());
put_request->set_key(key);
put_request->set_value(value);
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_put(put_request.release());
get_request.reset(new RangeRequest());
get_request->set_key(key);
req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("set")); std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("set"));
call->response_reader = stub_->AsyncTxn(&call->context,transaction.txn_request,&call->cq_);
call->response_reader = stub_->AsyncTxn(&call->context,txn_request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call.get()); call->response_reader->Finish(&call->reply, &call->status, (void*)call.get());
return Response::create(call); return Response::create(call);
@ -383,14 +243,9 @@ pplx::task<etcd::Response> etcd::Client::send_asyncput(std::string const & key,
pplx::task<etcd::Response> etcd::Client::send_asyncdelete(std::string const & key, bool recursive) pplx::task<etcd::Response> etcd::Client::send_asyncdelete(std::string const & key, bool recursive)
{ {
etcdv3::Transaction transaction(key);
//check if key is present transaction.init_compare(Compare::CompareResult::Compare_CompareResult_GREATER,
TxnRequest txn_request; Compare::CompareTarget::Compare_CompareTarget_VERSION);
Compare* compare = txn_request.add_compare();
compare->set_result(Compare::CompareResult::Compare_CompareResult_GREATER);
compare->set_target(Compare::CompareTarget::Compare_CompareTarget_VERSION);
compare->set_key(key);
compare->set_version(0);
std::string range_end(key); std::string range_end(key);
if(recursive) if(recursive)
@ -399,57 +254,11 @@ pplx::task<etcd::Response> etcd::Client::send_asyncdelete(std::string const & ke
range_end.back() = ascii+1; range_end.back() = ascii+1;
} }
//if success, get key, delete transaction.setup_delete_sequence(key, range_end, recursive);
std::unique_ptr<RangeRequest> get_request(new RangeRequest()); transaction.setup_delete_failure_operation(key, range_end, recursive);
get_request->set_key(key);
if(recursive)
{
get_request->set_range_end(range_end);
get_request->set_sort_target(RangeRequest::SortTarget::RangeRequest_SortTarget_KEY);
get_request->set_sort_order(RangeRequest::SortOrder::RangeRequest_SortOrder_ASCEND);
}
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::unique_ptr<DeleteRangeRequest> del_request(new DeleteRangeRequest());
del_request->set_key(key);
if(recursive)
{
del_request->set_range_end(range_end);
}
req_success = txn_request.add_success();
req_success->set_allocated_request_delete_range(del_request.release());
//if fail, get key, delete
get_request.reset(new RangeRequest());
get_request->set_key(key);
if(recursive)
{
get_request->set_range_end(range_end);
get_request->set_sort_target(RangeRequest::SortTarget::RangeRequest_SortTarget_KEY);
get_request->set_sort_order(RangeRequest::SortOrder::RangeRequest_SortOrder_ASCEND);
}
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
del_request.reset(new DeleteRangeRequest());
del_request->set_key(key);
if(recursive)
{
del_request->set_range_end(range_end);
}
req_failure = txn_request.add_failure();
req_failure->set_allocated_request_delete_range(del_request.release());
std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("delete")); std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("delete"));
call->response_reader = stub_->AsyncTxn(&call->context,transaction.txn_request,&call->cq_);
call->response_reader = stub_->AsyncTxn(&call->context,txn_request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call.get()); call->response_reader->Finish(&call->reply, &call->status, (void*)call.get());
return Response::create(call); return Response::create(call);
@ -457,82 +266,33 @@ pplx::task<etcd::Response> etcd::Client::send_asyncdelete(std::string const & ke
pplx::task<etcd::Response> etcd::Client::send_asyncrm_if(std::string const &key, std::string const &old_value) pplx::task<etcd::Response> etcd::Client::send_asyncrm_if(std::string const &key, std::string const &old_value)
{ {
etcdv3::Transaction transaction(key);
transaction.init_compare(old_value, Compare::CompareResult::Compare_CompareResult_EQUAL,
Compare::CompareTarget::Compare_CompareTarget_VALUE);
//check if key's value is equl to oldvalue transaction.setup_compare_and_delete_operation(key);
TxnRequest txn_request; transaction.setup_basic_failure_operation(key);
Compare* compare = txn_request.add_compare();
compare->set_result(Compare::CompareResult::Compare_CompareResult_EQUAL);
compare->set_target(Compare::CompareTarget::Compare_CompareTarget_VALUE);
compare->set_key(key);
compare->set_value(old_value);
//if success, get key, delete
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
get_request->set_key(key);
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::unique_ptr<DeleteRangeRequest> del_request(new DeleteRangeRequest());
del_request->set_key(key);
req_success = txn_request.add_success();
req_success->set_allocated_request_delete_range(del_request.release());
//if fail, get key
get_request.reset(new RangeRequest());
get_request->set_key(key);
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("compareAndDelete")); std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("compareAndDelete"));
call->response_reader = stub_->AsyncTxn(&call->context,transaction.txn_request,&call->cq_);
call->response_reader = stub_->AsyncTxn(&call->context,txn_request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call.get()); call->response_reader->Finish(&call->reply, &call->status, (void*)call.get());
return Response::create(call); return Response::create(call);
} }
pplx::task<etcd::Response> etcd::Client::send_asyncrm_if(std::string const &key, int old_index) pplx::task<etcd::Response> etcd::Client::send_asyncrm_if(std::string const &key, int old_index) {
{ etcdv3::Transaction transaction(key);
transaction.init_compare(old_index, Compare::CompareResult::Compare_CompareResult_EQUAL,
Compare::CompareTarget::Compare_CompareTarget_MOD);
//check if key's mod revision is equal to old index transaction.setup_compare_and_delete_operation(key);
TxnRequest txn_request; transaction.setup_basic_failure_operation(key);
Compare* compare = txn_request.add_compare();
compare->set_result(Compare::CompareResult::Compare_CompareResult_EQUAL);
compare->set_target(Compare::CompareTarget::Compare_CompareTarget_MOD);
compare->set_key(key);
compare->set_mod_revision(old_index);
std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("compareAndDelete"));
call->response_reader = stub_->AsyncTxn(&call->context,transaction.txn_request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call.get());
//if success, get key, delete return Response::create(call);
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
get_request->set_key(key);
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::unique_ptr<DeleteRangeRequest> del_request(new DeleteRangeRequest());
del_request->set_key(key);
req_success = txn_request.add_success();
req_success->set_allocated_request_delete_range(del_request.release());
//if fail, get key
get_request.reset(new RangeRequest());
get_request->set_key(key);
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
std::shared_ptr<etcdv3::AsyncTxnResponse> call(new etcdv3::AsyncTxnResponse("compareAndDelete"));
call->response_reader = stub_->AsyncTxn(&call->context,txn_request,&call->cq_);
call->response_reader->Finish(&call->reply, &call->status, (void*)call.get());
return Response::create(call);
} }

130
tst/EtcdTest.cpp
View File

@ -213,71 +213,71 @@ TEST_CASE("delete a directory")
} }
TEST_CASE("wait for a value change") //TEST_CASE("wait for a value change")
{ //{
etcd::Client etcd("http://127.0.0.1:2379"); // etcd::Client etcd("http://127.0.0.1:2379");
etcd.set("/test/key1", "42").wait(); // etcd.set("/test/key1", "42").wait();
//
pplx::task<etcd::Response> res = etcd.watch("/test/key1"); // pplx::task<etcd::Response> res = etcd.watch("/test/key1");
CHECK(!res.is_done()); // CHECK(!res.is_done());
sleep(1); // sleep(1);
CHECK(!res.is_done()); // CHECK(!res.is_done());
//
etcd.set("/test/key1", "43").get(); // etcd.set("/test/key1", "43").get();
sleep(1); // sleep(1);
REQUIRE(res.is_done()); // REQUIRE(res.is_done());
REQUIRE("set" == res.get().action()); // REQUIRE("set" == res.get().action());
CHECK("43" == res.get().value().as_string()); // CHECK("43" == res.get().value().as_string());
} //}
//
TEST_CASE("wait for a directory change") //TEST_CASE("wait for a directory change")
{ //{
etcd::Client etcd("http://127.0.0.1:4001"); // etcd::Client etcd("http://127.0.0.1:4001");
//
pplx::task<etcd::Response> res = etcd.watch("/test", true); // pplx::task<etcd::Response> res = etcd.watch("/test", true);
CHECK(!res.is_done()); // CHECK(!res.is_done());
sleep(1); // sleep(1);
CHECK(!res.is_done()); // CHECK(!res.is_done());
//
etcd.add("/test/key4", "44").wait(); // etcd.add("/test/key4", "44").wait();
sleep(1); // sleep(1);
REQUIRE(res.is_done()); // REQUIRE(res.is_done());
CHECK("create" == res.get().action()); // CHECK("create" == res.get().action());
CHECK("44" == res.get().value().as_string()); // CHECK("44" == res.get().value().as_string());
//
pplx::task<etcd::Response> res2 = etcd.watch("/test", true); // pplx::task<etcd::Response> res2 = etcd.watch("/test", true);
CHECK(!res2.is_done()); // CHECK(!res2.is_done());
sleep(1); // sleep(1);
CHECK(!res2.is_done()); // CHECK(!res2.is_done());
//
etcd.set("/test/key4", "45").wait(); // etcd.set("/test/key4", "45").wait();
REQUIRE(res2.is_done()); // REQUIRE(res2.is_done());
CHECK("set" == res2.get().action()); // CHECK("set" == res2.get().action());
CHECK("45" == res2.get().value().as_string()); // CHECK("45" == res2.get().value().as_string());
} //}
//
TEST_CASE("watch changes in the past") //TEST_CASE("watch changes in the past")
{ //{
etcd::Client etcd("http://127.0.0.1:4001"); // etcd::Client etcd("http://127.0.0.1:4001");
//
int index = etcd.set("/test/key1", "42").get().index(); // int index = etcd.set("/test/key1", "42").get().index();
//
etcd.set("/test/key1", "43").wait(); // etcd.set("/test/key1", "43").wait();
etcd.set("/test/key1", "44").wait(); // etcd.set("/test/key1", "44").wait();
etcd.set("/test/key1", "45").wait(); // etcd.set("/test/key1", "45").wait();
//
etcd::Response res = etcd.watch("/test/key1", ++index).get(); // etcd::Response res = etcd.watch("/test/key1", ++index).get();
CHECK("set" == res.action()); // CHECK("set" == res.action());
CHECK("43" == res.value().as_string()); // CHECK("43" == res.value().as_string());
//
res = etcd.watch("/test/key1", ++index).get(); // res = etcd.watch("/test/key1", ++index).get();
CHECK("set" == res.action()); // CHECK("set" == res.action());
CHECK("44" == res.value().as_string()); // CHECK("44" == res.value().as_string());
//
res = etcd.watch("/test", ++index, true).get(); // res = etcd.watch("/test", ++index, true).get();
CHECK("set" == res.action()); // CHECK("set" == res.action());
CHECK("45" == res.value().as_string()); // CHECK("45" == res.value().as_string());
} //}
TEST_CASE("cleanup") TEST_CASE("cleanup")

43
v3/include/Transaction.hpp Normal file
View File

@ -0,0 +1,43 @@
/*
* Transaction.hpp
*
* Created on: Jun 21, 2016
* Author: ubuntu
*/
#ifndef V3_SRC_TRANSACTION_HPP_
#define V3_SRC_TRANSACTION_HPP_
#include <grpc++/grpc++.h>
#include "proto/rpc.grpc.pb.h"
#include <string>
namespace etcdv3 {
class Transaction {
public:
Transaction();
Transaction(std::string const&);
virtual ~Transaction();
void init_compare(etcdserverpb::Compare::CompareResult, etcdserverpb::Compare::CompareTarget);
void init_compare(std::string const &, etcdserverpb::Compare::CompareResult, etcdserverpb::Compare::CompareTarget);
void init_compare(int, etcdserverpb::Compare::CompareResult, etcdserverpb::Compare::CompareTarget);
void setup_basic_failure_operation(std::string const &key);
void setup_set_failure_operation(std::string const &key, std::string const &value);
void setup_basic_create_sequence(std::string const &key, std::string const &value);
void setup_compare_and_swap_sequence(std::string const &valueToSwap);
void setup_delete_sequence(std::string const &key, std::string const &range_end, bool recursive);
void setup_delete_failure_operation(std::string const &key, std::string const &range_end, bool recursive);
void setup_compare_and_delete_operation(std::string const& key);
etcdserverpb::TxnRequest txn_request;
private:
std::string key;
};
} /* namespace etcdv3 */
#endif /* V3_SRC_TRANSACTION_HPP_ */

178
v3/src/Transaction.cpp Normal file
View File

@ -0,0 +1,178 @@
/*
* Transaction.cpp
*
* Created on: Jun 21, 2016
* Author: ubuntu
*/
#include "v3/include/Transaction.hpp"
using etcdserverpb::Compare;
using etcdserverpb::RangeRequest;
using etcdserverpb::PutRequest;
using etcdserverpb::RequestOp;
using etcdserverpb::DeleteRangeRequest;
etcdv3::Transaction::Transaction() {
// TODO Auto-generated constructor stub
}
etcdv3::Transaction::Transaction(const std::string& key) : key(key) {
}
void etcdv3::Transaction::init_compare(Compare::CompareResult result, Compare::CompareTarget target){
Compare* compare = txn_request.add_compare();
compare->set_result(result);
compare->set_target(target);
compare->set_key(key);
compare->set_version(0);
}
void etcdv3::Transaction::init_compare(std::string const& old_value, Compare::CompareResult result, Compare::CompareTarget target){
Compare* compare = txn_request.add_compare();
compare->set_result(result);
compare->set_target(target);
compare->set_key(key);
compare->set_value(old_value);
}
void etcdv3::Transaction::init_compare(int old_index, Compare::CompareResult result, Compare::CompareTarget target){
Compare* compare = txn_request.add_compare();
compare->set_result(result);
compare->set_target(target);
compare->set_key(key);
compare->set_mod_revision(old_index);
}
void etcdv3::Transaction::setup_basic_failure_operation(std::string const& key) {
std::unique_ptr<RangeRequest> get_request(new RangeRequest()); //then this can be just a raw pointer
get_request->set_key(key);
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release()); //why not get()?
}
void etcdv3::Transaction::setup_set_failure_operation(std::string const &key, std::string const &value) {
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
get_request->set_key(key);
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
std::unique_ptr<PutRequest> put_request(new PutRequest());
put_request->set_key(key);
put_request->set_value(value);
req_failure = txn_request.add_failure();
req_failure->set_allocated_request_put(put_request.release());
get_request.reset(new RangeRequest());
get_request->set_key(key);
req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
}
void etcdv3::Transaction::setup_delete_failure_operation(std::string const &key, std::string const &range_end, bool recursive) {
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
std::unique_ptr<DeleteRangeRequest> del_request(new DeleteRangeRequest());
get_request.reset(new RangeRequest());
get_request->set_key(key);
if(recursive)
{
get_request->set_range_end(range_end);
get_request->set_sort_target(RangeRequest::SortTarget::RangeRequest_SortTarget_KEY);
get_request->set_sort_order(RangeRequest::SortOrder::RangeRequest_SortOrder_ASCEND);
}
RequestOp* req_failure = txn_request.add_failure();
req_failure->set_allocated_request_range(get_request.release());
del_request.reset(new DeleteRangeRequest());
del_request->set_key(key);
if(recursive)
{
del_request->set_range_end(range_end);
}
req_failure = txn_request.add_failure();
req_failure->set_allocated_request_delete_range(del_request.release());
}
/**
* add key and then get new value of key
*/
void etcdv3::Transaction::setup_basic_create_sequence(std::string const& key, std::string const& value) {
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
std::unique_ptr<PutRequest> put_request(new PutRequest());
put_request->set_key(key);
put_request->set_value(value);
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_put(put_request.release());
get_request->set_key(key);
req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
}
/**
* get key value then modify and get new value
*/
void etcdv3::Transaction::setup_compare_and_swap_sequence(std::string const& value) {
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
get_request.reset(new RangeRequest());
get_request->set_key(key);
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::unique_ptr<PutRequest> put_request(new PutRequest());
put_request->set_key(key);
put_request->set_value(value);
req_success = txn_request.add_success();
req_success->set_allocated_request_put(put_request.release());
get_request.reset(new RangeRequest());
get_request->set_key(key);
req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
}
void etcdv3::Transaction::setup_delete_sequence(std::string const &key, std::string const &range_end, bool recursive) {
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
get_request->set_key(key);
if(recursive)
{
get_request->set_range_end(range_end);
get_request->set_sort_target(RangeRequest::SortTarget::RangeRequest_SortTarget_KEY);
get_request->set_sort_order(RangeRequest::SortOrder::RangeRequest_SortOrder_ASCEND);
}
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::unique_ptr<DeleteRangeRequest> del_request(new DeleteRangeRequest());
del_request->set_key(key);
if(recursive)
{
del_request->set_range_end(range_end);
}
req_success = txn_request.add_success();
req_success->set_allocated_request_delete_range(del_request.release());
}
void etcdv3::Transaction::setup_compare_and_delete_operation(std::string const& key) {
std::unique_ptr<RangeRequest> get_request(new RangeRequest());
get_request->set_key(key);
RequestOp* req_success = txn_request.add_success();
req_success->set_allocated_request_range(get_request.release());
std::unique_ptr<DeleteRangeRequest> del_request(new DeleteRangeRequest());
del_request->set_key(key);
req_success = txn_request.add_success();
req_success->set_allocated_request_delete_range(del_request.release());
}
etcdv3::Transaction::~Transaction() {
// TODO Auto-generated destructor stub
}