From 38366fc5c37cba64074c54418b4c12c8d723a92c Mon Sep 17 00:00:00 2001
From: Tao He <linzhu.ht@alibaba-inc.com>
Date: Mon, 15 Feb 2021 00:13:09 +0800
Subject: [PATCH] Handle watching on compacted revisions cases.

Fixes #43.

Signed-off-by: Tao He <linzhu.ht@alibaba-inc.com>
---
 src/v3/AsyncWatchAction.cpp   | 15 ++++++++++++++-
 src/v3/AsyncWatchResponse.cpp |  5 +++++
 2 files changed, 19 insertions(+), 1 deletion(-)

diff --git a/src/v3/AsyncWatchAction.cpp b/src/v3/AsyncWatchAction.cpp
index f4e2ae5..8b1b9cd 100644
--- a/src/v3/AsyncWatchAction.cpp
+++ b/src/v3/AsyncWatchAction.cpp
@@ -63,7 +63,11 @@ void etcdv3::AsyncWatchAction::waitForResponse()
     }
     if(got_tag == (void*)this) // read tag
     {
-      if ((reply.created() && reply.header().revision() < parameters.revision) ||
+      if (reply.canceled()) {
+        isCancelled = true;
+        cq_.Shutdown();
+      }
+      else if ((reply.created() && reply.header().revision() < parameters.revision) ||
           reply.events_size() > 0) {
         // we stop watch under two conditions:
         //
@@ -125,6 +129,15 @@ void etcdv3::AsyncWatchAction::waitForResponse(std::function<void(etcd::Response
     }
     else if(got_tag == (void*)this) // read tag
     {
+      if (reply.canceled()) {
+        isCancelled = true;
+        cq_.Shutdown();
+        if (reply.compact_revision() != 0) {
+          callback(etcd::Response(grpc::StatusCode::OUT_OF_RANGE /* error code */,
+                                  "required revision has been compacted"));
+        }
+        break;
+      }
       if(reply.events_size())
       {
         // for the callback case, we don't stop immediately if watching for a future revison,
diff --git a/src/v3/AsyncWatchResponse.cpp b/src/v3/AsyncWatchResponse.cpp
index c06b116..8e2da1a 100644
--- a/src/v3/AsyncWatchResponse.cpp
+++ b/src/v3/AsyncWatchResponse.cpp
@@ -3,6 +3,11 @@
 
 void etcdv3::AsyncWatchResponse::ParseResponse(WatchResponse& reply)
 {
+  if (reply.canceled() && reply.compact_revision() != 0) {
+    error_code=grpc::StatusCode::OUT_OF_RANGE;
+    error_message="required revision has been compacted";
+    return;
+  }
   index = reply.header().revision();
   for (auto const &e: reply.events()) {
     events.emplace_back(e);