add etcd watch implementation

src/main/java/com/coreos/jetcd/EtcdClient.java

@@ -34,6 +34,7 @@ public class EtcdClient {
     private final Supplier<EtcdMaintenance>       maintenanceClient;
     private final Supplier<EtcdCluster>           clusterClient;
     private final Supplier<EtcdLease>             leaseClient;
+    private final Supplier<EtcdWatch>             watchClient;
 
     public EtcdClient(EtcdClientBuilder builder) throws ConnectException, AuthFailedException {
         this(Optional.empty(), builder);
@@ -62,6 +63,7 @@ private EtcdClient(Optional<ManagedChannelBuilder<?>> channelBuilder, EtcdClient
         this.maintenanceClient = Suppliers.memoize(() -> new EtcdMaintenanceImpl(channel, token));
         this.clusterClient =  Suppliers.memoize(() -> new EtcdClusterImpl(channel, token));
         this.leaseClient =  Suppliers.memoize(() -> new EtcdLeaseImpl(channel, token));
+        this.watchClient =  Suppliers.memoize(() -> new EtcdWatchImpl(channel, token));
     }
 
     // ************************
@@ -88,6 +90,10 @@ public EtcdLease getLeaseClient() {
         return this.leaseClient.get();
     }
 
+    public EtcdWatch getWatchClient() {
+        return this.watchClient.get();
+    }
+
     public void close() {
         channel.shutdownNow();
     }

src/main/java/com/coreos/jetcd/EtcdUtil.java

@@ -0,0 +1,80 @@
+package com.coreos.jetcd;
+
+import com.coreos.jetcd.api.Event;
+import com.coreos.jetcd.api.ResponseHeader;
+import com.coreos.jetcd.data.ByteSequence;
+import com.coreos.jetcd.data.EtcdHeader;
+import com.coreos.jetcd.data.KeyValue;
+import com.coreos.jetcd.watch.WatchEvent;
+import com.google.protobuf.ByteString;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * This util is to convert api class to client class.
+ */
+class EtcdUtil {
+
+    private EtcdUtil() {
+    }
+
+    /**
+     * convert ByteSequence to ByteString
+     */
+    protected static ByteString byteStringFromByteSequence(ByteSequence byteSequence) {
+        return ByteString.copyFrom(byteSequence.getBytes());
+    }
+
+    /**
+     * convert ByteString to ByteSequence
+     * @return
+     */
+    protected static ByteSequence byteSequceFromByteString(ByteString byteString) {
+        return ByteSequence.fromBytes(byteString.toByteArray());
+    }
+
+    /**
+     * convert API KeyValue to etcd client KeyValue
+     */
+    protected static KeyValue apiToClientKV(com.coreos.jetcd.api.KeyValue keyValue) {
+        return new KeyValue(
+                byteSequceFromByteString(keyValue.getKey()),
+                byteSequceFromByteString(keyValue.getValue()),
+                keyValue.getCreateRevision(),
+                keyValue.getModRevision(),
+                keyValue.getVersion(),
+                keyValue.getLease());
+    }
+
+    /**
+     * convert API watch event to etcd client event
+     */
+    protected static WatchEvent apiToClientEvent(Event event) {
+        WatchEvent.EventType eventType = WatchEvent.EventType.UNRECOGNIZED;
+        switch (event.getType()) {
+            case DELETE:
+                eventType = WatchEvent.EventType.DELETE;
+                break;
+            case PUT:
+                eventType = WatchEvent.EventType.PUT;
+                break;
+        }
+        return new WatchEvent(apiToClientKV(event.getKv()), apiToClientKV(event.getPrevKv()), eventType);
+    }
+
+    protected static List<WatchEvent> apiToClientEvents(List<Event> events) {
+        List<WatchEvent> watchEvents = new ArrayList<>();
+        for (Event event : events) {
+            watchEvents.add(apiToClientEvent(event));
+        }
+        return watchEvents;
+    }
+
+    /**
+     * convert API response header to self defined header
+     */
+    protected static EtcdHeader apiToClientHeader(ResponseHeader header, long compactRevision) {
+        return new EtcdHeader(header.getClusterId(), header.getMemberId(), header.getRevision(), header.getRaftTerm(), compactRevision);
+    }
+}

src/main/java/com/coreos/jetcd/EtcdWatch.java

@@ -0,0 +1,74 @@
+package com.coreos.jetcd;
+
+import com.coreos.jetcd.data.ByteSequence;
+import com.coreos.jetcd.data.EtcdHeader;
+import com.coreos.jetcd.options.WatchOption;
+import com.coreos.jetcd.watch.WatchEvent;
+
+import java.io.Closeable;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Interface of watch client
+ */
+public interface EtcdWatch {
+
+
+    /**
+     * Watch watches on a key or prefix. The watched events will be called by onWatch.
+     * If the watch is slow or the required rev is compacted, the watch request
+     * might be canceled from the server-side and the onCreateFailed will be called.
+     *
+     * @param key         the key subscribe
+     * @param watchOption key option
+     * @param callback    call back
+     * @return ListenableFuture watcher
+     */
+    CompletableFuture<Watcher> watch(ByteSequence key, WatchOption watchOption, WatchCallback callback);
+
+    interface Watcher extends Closeable {
+
+        /**
+         * get watcher id
+         *
+         * @return id
+         */
+        long getWatchID();
+
+        long getLastRevision();
+
+        ByteSequence getKey();
+
+        boolean isResuming();
+
+        /**
+         * get the watch option
+         *
+         * @return watch option
+         */
+        WatchOption getWatchOption();
+
+        /**
+         * cancel the watcher
+         *
+         * @return cancel result
+         */
+        CompletableFuture<Boolean> cancel();
+    }
+
+    interface WatchCallback {
+
+        /**
+         * onWatch will be called when watcher receive any events
+         *
+         * @param events received events
+         */
+        void onWatch(EtcdHeader header, List<WatchEvent> events);
+
+        /**
+         * onResuming will be called when the watcher is on resuming.
+         */
+        void onResuming();
+    }
+}