import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
@Override
protected void finishedPayload(byte[] pl) throws IOException {
  if (keyLen > 0) {
    final byte[] keyBytes = new byte[keyLen];
    final byte[] data = new byte[pl.length - keyLen];
    System.arraycopy(pl, 0, keyBytes, 0, keyLen);
    System.arraycopy(pl, keyLen, data, 0, pl.length - keyLen);
    Callback cb = (Callback) getCallback();
    cb.gotStat(new String(keyBytes, "UTF-8"), new String(data, "UTF-8"));
  } else {
    OperationStatus status = getStatusForErrorCode(errorCode, pl);
    getCallback().receivedStatus(status);
    transitionState(OperationState.COMPLETE);
  }
  resetInput();
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
protected void finishedPayload(byte[] pl) throws IOException {
  OperationStatus status = getStatusForErrorCode(errorCode, pl);

  if (status == null) {
    handleError(OperationErrorType.SERVER, new String(pl));
  } else if (errorCode == SUCCESS) {
    decodePayload(pl);
    transitionState(OperationState.COMPLETE);
  } else if (errorCode == ERR_NOT_MY_VBUCKET
      && !getState().equals(OperationState.COMPLETE)) {
    transitionState(OperationState.RETRY);
  } else {
    getCallback().receivedStatus(status);
    transitionState(OperationState.COMPLETE);
  }
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
/**
 * Wait for the queues to die down.
 *
 * @param timeout the amount of time time for shutdown
 * @param unit the TimeUnit for the timeout
 * @return result of the request for the wait
 * @throws IllegalStateException in the rare circumstance where queue is too
 *           full to accept any more requests
 */
public boolean waitForQueues(long timeout, TimeUnit unit) {
  CountDownLatch blatch = broadcastOp(new BroadcastOpFactory() {
    public Operation newOp(final MemcachedNode n,
        final CountDownLatch latch) {
      return opFact.noop(new OperationCallback() {
        public void complete() {
          latch.countDown();
        }

        public void receivedStatus(OperationStatus s) {
          // Nothing special when receiving status, only
          // necessary to complete the interface
        }
      });
    }
  }, conn.getLocator().getAll(), false);
  try {
    // XXX: Perhaps IllegalStateException should be caught here
    // and the check retried.
    return blatch.await(timeout, unit);
  } catch (InterruptedException e) {
    throw new RuntimeException("Interrupted waiting for queues", e);
  }
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
private static void sendSpyMemcached(final MemcachedClient client,
                                     final List<String> keys,
                                     final Map<String, String> expectedMap,
                                     final ProgressMeter meter,
                                     final ScheduledExecutorService backoffExecutor) {
  final long start = System.nanoTime();
  final BulkFuture<Map<String, Object>> future = client.asyncGetBulk(keys);
  future.addListener(getFuture -> {
    final OperationStatus status = getFuture.getStatus();
    if (status.isSuccess()) {
      final Map<String, String> response = (Map<String, String>) getFuture.get();
      final long end = System.nanoTime();
      final long latency = end - start;
      meter.inc(keys.size(), latency);
      if (!expectedMap.equals(response)) {
        throw new AssertionError("expected: " + expectedMap + ", got: " + response);
      }
      sendSpyMemcached(client, keys, expectedMap, meter, backoffExecutor);
    } else {
      System.err.println("failure!");
    }
  });
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
private static OperationStatus getStatusWithretriesOnTimeout(
    OperationFuture<Boolean> rv, int tries) {
int i = 0;
boolean shouldRetry = false;
do {
    try {
	shouldRetry = false;
	return rv.getStatus();
    } catch (Exception exception) {
	if (exception.getCause() instanceof CheckedOperationTimeoutException
		|| exception instanceof CheckedOperationTimeoutException) {
	    log.warn("Try[" + (tries - i) + "] Wait for response. Of:"
		    + rv);
	    shouldRetry = true;
	}
    }

} while (++i < tries && shouldRetry);

return rv.getStatus();
   }
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
@Override
protected OperationStatus getStatusForErrorCode(int errCode, byte[] errPl) {
	OperationStatus rv=null;
	switch(errCode) {
		case ERR_EXISTS:
			rv=EXISTS_STATUS;
			break;
		case ERR_NOT_FOUND:
			rv=NOT_FOUND_STATUS;
			break;
		case ERR_NOT_STORED:
			rv=NOT_FOUND_STATUS;
			break;
	}
	return rv;
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
/**
 * <result value>\r\n
 */
public void handleLine(String line) {

	OperationStatus status = null;

	/* ENABLE_REPLICATION if */
	if (line.equals("SWITCHOVER") || line.equals("REPL_SLAVE")) {
		receivedMoveOperations(line);
		return;
	}

	/* ENABLE_REPLICATION end */
	try {
		Long.valueOf(line);
		getCallback().receivedStatus(new OperationStatus(true, line));
	} catch (NumberFormatException e) {
		status = matchStatus(line, NOT_FOUND, NOT_FOUND_ELEMENT, TYPE_MISMATCH, BKEY_MISMATCH,
				UNREADABLE, OVERFLOWED, OUT_OF_RANGE);

		getLogger().debug(status);
		getCallback().receivedStatus(status);
	}

	transitionState(OperationState.COMPLETE);
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
public void handleLine(String line) {
	getLogger().debug("Got line %s", line);

	if (line.startsWith("VALUE ")) {
		readKey(line);
		setReadType(OperationReadType.DATA);
	} else {
		OperationStatus status = matchStatus(line, END);

		getLogger().debug(status);
		getCallback().receivedStatus(status);

		transitionState(OperationState.COMPLETE);
		return;
	}
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
@Override
public void handleLine(String line) {
	/* ENABLE_REPLICATION if */
	if (line.equals("SWITCHOVER") || line.equals("REPL_SLAVE")) {
		receivedMoveOperations(line);
	}

	/* ENABLE_REPLICATION end */
	OperationStatus status=null;
	try {
		Long.valueOf(line);
		getCallback().receivedStatus(new OperationStatus(true, line));
	} catch (NumberFormatException e) {
		status = matchStatus(line, NOT_FOUND, TYPE_MISMATCH);
		getCallback().receivedStatus(status);
	}
	transitionState(OperationState.COMPLETE);
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
Future<Boolean> asyncStore(StoreType storeType, String key,
		int exp, CachedData co) {
	final CountDownLatch latch=new CountDownLatch(1);
	final OperationFuture<Boolean> rv=new OperationFuture<Boolean>(latch,
			operationTimeout);
	Operation op=opFact.store(storeType, key, co.getFlags(),
			exp, co.getData(), new OperationCallback() {
		public void receivedStatus(OperationStatus val) {
			rv.set(val.isSuccess());
		}
		public void complete() {
			latch.countDown();
		}});
	rv.setOperation(op);
	addOp(key, op);
	return rv;
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
@Override
public CollectionFuture<Boolean> asyncSetAttr(String key,
		Attributes attrs) {
	final CountDownLatch latch = new CountDownLatch(1);
	final CollectionFuture<Boolean> rv = new CollectionFuture<Boolean>(
			latch, operationTimeout);
	Operation op = opFact.setAttr(key, attrs, new OperationCallback() {
		public void receivedStatus(OperationStatus status) {
			if (status instanceof CollectionOperationStatus) {
				rv.set(status.isSuccess(), (CollectionOperationStatus) status);
			} else {
				getLogger().warn("Unhandled state: " + status);
				rv.set(status.isSuccess(), new CollectionOperationStatus(status));
			}
		}
		public void complete() {
			latch.countDown();
		}
	});
	rv.setOperation(op);
	addOp(key, op);
	return rv;
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
@RequestMapping("/play/{cinema}/{show}")
public ResponseEntity<String> play(@PathVariable String cinema, @PathVariable String show) throws Exception {
    Show s = new Show();
    s.setCinema_id("cinema::" + cinema);
    s.setDuration(TimeUnit.HOURS.toSeconds(2));
    s.setName(show);
    s.setStart(System.currentTimeMillis() / 1000L);

    String showId = s.getCinema_id() + "::" + nextShowId(s.getCinema_id());
    OperationFuture<Boolean> future = db.add(showId, mapper.writeValueAsString(s));
    future.get();
    OperationStatus status = future.getStatus();
    if (status.isSuccess()) {
        return new ResponseEntity<String>(HttpStatus.CREATED);
    } else {
        LOGGER.warn("/play/{}/{} failed because of {}", cinema, show, status.getMessage());
        return new ResponseEntity<String>(HttpStatus.INTERNAL_SERVER_ERROR);
    }
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
private <T> Future<Boolean> asyncCat(
		ConcatenationType catType, long cas, String key,
		T value, Transcoder<T> tc) {
	CachedData co=tc.encode(value);
	final CountDownLatch latch=new CountDownLatch(1);
	final OperationFuture<Boolean> rv=new OperationFuture<Boolean>(latch,
			operationTimeout);
	Operation op=opFact.cat(catType, cas, key, co.getData(),
			new OperationCallback() {
		public void receivedStatus(OperationStatus val) {
			rv.set(val.isSuccess());
		}
		public void complete() {
			latch.countDown();
		}});
	rv.setOperation(op);
	addOp(key, op);
	return rv;
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
/**
 * Asynchronous CAS operation.
 *
 * @param <T>
 * @param key the key
 * @param casId the CAS identifier (from a gets operation)
 * @param exp the expiration of this object
 * @param value the new value
 * @param tc the transcoder to serialize and unserialize the value
 * @return a future that will indicate the status of the CAS
 * @throws IllegalStateException in the rare circumstance where queue
 *         is too full to accept any more requests
 */
public <T> Future<CASResponse> asyncCAS(String key, long casId, int exp, T value,
		Transcoder<T> tc) {
	CachedData co=tc.encode(value);
	final CountDownLatch latch=new CountDownLatch(1);
	final OperationFuture<CASResponse> rv=new OperationFuture<CASResponse>(
			latch, operationTimeout);
	Operation op=opFact.cas(StoreType.set, key, casId, co.getFlags(), exp,
			co.getData(), new OperationCallback() {
				public void receivedStatus(OperationStatus val) {
					if(val instanceof CASOperationStatus) {
						rv.set(((CASOperationStatus)val).getCASResponse());
					} else if(val instanceof CancelledOperationStatus) {
						rv.set(CASResponse.CANCELED);
					} else {
						rv.set(CASResponse.UNDEFINED);
					}
				}
				public void complete() {
					latch.countDown();
				}});
	rv.setOperation(op);
	addOp(key, op);
	return rv;
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
/**
 * Get the versions of all of the connected memcacheds.
 *
 * @return a Map of SocketAddress to String for connected servers
 * @throws IllegalStateException in the rare circumstance where queue
 *         is too full to accept any more requests
 */
public Map<SocketAddress, String> getVersions() {
	final Map<SocketAddress, String>rv=
		new ConcurrentHashMap<SocketAddress, String>();

	CountDownLatch blatch = broadcastOp(new BroadcastOpFactory(){
		public Operation newOp(final MemcachedNode n,
				final CountDownLatch latch) {
			final SocketAddress sa=n.getSocketAddress();
			return opFact.version(
					new OperationCallback() {
						public void receivedStatus(OperationStatus s) {
							rv.put(sa, s.getMessage());
						}
						public void complete() {
							latch.countDown();
						}
					});
		}});
	try {
		blatch.await(operationTimeout, TimeUnit.MILLISECONDS);
	} catch (InterruptedException e) {
		throw new RuntimeException("Interrupted waiting for versions", e);
	}
	return rv;
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
private long mutate(Mutator m, String key, int by, long def, int exp) {
	final AtomicLong rv=new AtomicLong();
	final CountDownLatch latch=new CountDownLatch(1);
	addOp(key, opFact.mutate(m, key, by, def, exp, new OperationCallback() {
				public void receivedStatus(OperationStatus s) {
					// XXX:  Potential abstraction leak.
					// The handling of incr/decr in the binary protocol
					// Allows us to avoid string processing.
					rv.set(new Long(s.isSuccess()?s.getMessage():"-1"));
				}
				public void complete() {
					latch.countDown();
				}}));
	try {
		if (!latch.await(operationTimeout, TimeUnit.MILLISECONDS)) {
			throw new OperationTimeoutException(
				"Mutate operation timed out, unable to modify counter ["
					+ key + "]");
		}
	} catch (InterruptedException e) {
		throw new RuntimeException("Interrupted", e);
	}
	getLogger().debug("Mutation returned %s", rv);
	return rv.get();
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
private Future<Long> asyncMutate(Mutator m, String key, int by, long def,
		int exp) {
	final CountDownLatch latch = new CountDownLatch(1);
	final OperationFuture<Long> rv = new OperationFuture<Long>(
			latch, operationTimeout);
	Operation op = addOp(key, opFact.mutate(m, key, by, def, exp,
			new OperationCallback() {
		public void receivedStatus(OperationStatus s) {
			rv.set(new Long(s.isSuccess() ? s.getMessage() : "-1"));
		}
		public void complete() {
			latch.countDown();
		}
	}));
	rv.setOperation(op);
	return rv;
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
/**
 * Delete the given key from the cache.
 *
 * @param key the key to delete
 * @return whether or not the operation was performed
 * @throws IllegalStateException in the rare circumstance where queue
 *         is too full to accept any more requests
 */
public Future<Boolean> delete(String key) {
	final CountDownLatch latch=new CountDownLatch(1);
	final OperationFuture<Boolean> rv=new OperationFuture<Boolean>(latch,
		operationTimeout);
	DeleteOperation op=opFact.delete(key,
			new OperationCallback() {
				public void receivedStatus(OperationStatus s) {
					rv.set(s.isSuccess());
				}
				public void complete() {
					latch.countDown();
				}});
	rv.setOperation(op);
	addOp(key, op);
	return rv;
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
/**
 * Wait for the queues to die down.
 *
 * @param timeout the amount of time time for shutdown
 * @param unit the TimeUnit for the timeout
 * @return result of the request for the wait
 * @throws IllegalStateException in the rare circumstance where queue
 *         is too full to accept any more requests
 */
public boolean waitForQueues(long timeout, TimeUnit unit) {
	CountDownLatch blatch = broadcastOp(new BroadcastOpFactory(){
		public Operation newOp(final MemcachedNode n,
				final CountDownLatch latch) {
			return opFact.noop(
					new OperationCallback() {
						public void complete() {
							latch.countDown();
						}
						public void receivedStatus(OperationStatus s) {
							// Nothing special when receiving status, only
							// necessary to complete the interface
						}
					});
		}}, conn.getLocator().getAll(), false);
	try {
		// XXX:  Perhaps IllegalStateException should be caught here
		// and the check retried.
		return blatch.await(timeout, unit);
	} catch (InterruptedException e) {
		throw new RuntimeException("Interrupted waiting for queues", e);
	}
}
 

import net.spy.memcached.ops.OperationStatus; //导入依赖的package包/类
@Override
protected void setUp() throws Exception {
	super.setUp();
	ofact = getOperationFactory();
	genericCallback = new OperationCallback() {
		public void complete() {
			fail("Unexpected invocation");
		}
		public void receivedStatus(OperationStatus status) {
			fail("Unexpected status:  " + status);
		}
	};

	testData = new byte[64];
	new Random().nextBytes(testData);
}