SPY-169: Also clone observe calls on redistribute.

Motivation
----------
Observe calls also need to be cloned when redistributed, for example if a socket
gets closed or the op stil waits for an auth latch.

Modifications
-------------
The observe operation now gets cloned similar to get, gets and so on.

Result
------
No assert error is raised if an observe needs to be cloned.

Change-Id: I6269a1d7fb756855f94caa2fe0495133fe1a83b0
Reviewed-on: http://review.couchbase.org/36698
Reviewed-by: Matt Ingenthron <matt@couchbase.com>
Tested-by: Michael Nitschinger <michael.nitschinger@couchbase.com>

Author: daschl

src/main/java/net/spy/memcached/ops/BaseOperationFactory.java

@@ -23,12 +23,12 @@
 
 package net.spy.memcached.ops;
 
-import java.util.ArrayList;
-import java.util.Collection;
-
 import net.spy.memcached.MemcachedNode;
 import net.spy.memcached.OperationFactory;
 
+import java.util.ArrayList;
+import java.util.Collection;
+
 /**
  * Base class for operation factories.
  *
@@ -93,6 +93,10 @@ public Collection<Operation> clone(KeyedOperation op) {
       GetAndTouchOperation gt = (GetAndTouchOperation) op;
       rv.add(getAndTouch(first(gt.getKeys()), gt.getExpiration(),
         (GetAndTouchOperation.Callback) gt.getCallback()));
+    } else if (op instanceof ObserveOperation) {
+      ObserveOperation oo = (ObserveOperation) op;
+      rv.add(observe(first(oo.getKeys()), oo.getCasValue(), oo.getIndex(),
+        (ObserveOperation.Callback) oo.getCallback()));
     } else {
       assert false : "Unhandled operation type: " + op.getClass();
     }

src/main/java/net/spy/memcached/ops/ObserveOperation.java

@@ -29,6 +29,17 @@
  * Observe operation.
  */
 public interface ObserveOperation extends KeyedOperation {
+
+  /**
+   * Get the CAS value advised for this operation.
+   */
+  long getCasValue();
+
+  /**
+   * Get the index advised for this operation.
+   */
+  int getIndex();
+
   /**
    * Operation callback for the Observe request.
    */

src/main/java/net/spy/memcached/protocol/binary/ObserveOperationImpl.java

@@ -67,4 +67,14 @@ protected void decodePayload(byte[] pl) {
         getHandlingNode(), r);
     getCallback().receivedStatus(STATUS_OK);
   }
+
+  @Override
+  public long getCasValue() {
+    return cas;
+  }
+
+  @Override
+  public int getIndex() {
+    return index;
+  }
 }