import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
private static String
createMessage(String message, Collection<Operation> ops) {
  StringBuilder rv = new StringBuilder(message);
  rv.append(" - failing node");
  rv.append(ops.size() == 1 ? ": " : "s: ");
  boolean first = true;
  for (Operation op : ops) {
    if (first) {
      first = false;
    } else {
      rv.append(", ");
    }
    MemcachedNode node = op == null ? null : op.getHandlingNode();
    rv.append(node == null ? "<unknown>" : node.getSocketAddress());
  }
  return rv.toString();
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
/**
 * Returns the socket address of a given MemcachedNode.
 *
 * @param node The node which we're interested in
 * @return String the socket address of that node.
 */
protected String getSocketAddressForNode(MemcachedNode node) {
  // Using the internal map retrieve the socket addresses
  // for given nodes.
  // I'm aware that this code is inherently thread-unsafe as
  // I'm using a HashMap implementation of the map, but the worst
  // case ( I believe) is we're slightly in-efficient when
  // a node has never been seen before concurrently on two different
  // threads, so it the socketaddress will be requested multiple times!
  // all other cases should be as fast as possible.
  String result = socketAddresses.get(node);
  if (result == null) {
    result = String.valueOf(node.getSocketAddress());
    if (result.startsWith("/")) {
      result = result.substring(1);
    }
    socketAddresses.put(node, result);
  }
  return result;
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
@Test
public void testSuccessful() {
    Long timeout = 1000L;
    TimeUnit unit = TimeUnit.MILLISECONDS;

    MockOperationFuture future = mock(MockOperationFuture.class);
    MockOperation operation = mock(MockOperation.class);

    try {
        when(operation.getException()).thenReturn(null);
        when(operation.isCancelled()).thenReturn(false);
        when(future.__getTraceObject1()).thenReturn(operation);

        MemcachedNode node = getMockMemcachedNode();
        when(operation.getHandlingNode()).thenReturn(node);

        interceptor.before(future, new Object[] { timeout, unit });
        interceptor.after(future, new Object[] { timeout, unit }, null, null);
    } catch (Exception e) {
        fail(e.getMessage());
    }
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
@Test
public void testTimeoutException() {
    Long timeout = 1000L;
    TimeUnit unit = TimeUnit.MILLISECONDS;

    MockOperationFuture future = mock(MockOperationFuture.class);
    MockOperation operation = mock(MockOperation.class);

    try {
        OperationException exception = new OperationException(OperationErrorType.GENERAL, "timed out");
        when(operation.getException()).thenReturn(exception);
        when(operation.isCancelled()).thenReturn(true);
        when(future.__getTraceObject1()).thenReturn(operation);

        MemcachedNode node = getMockMemcachedNode();
        when(operation.getHandlingNode()).thenReturn(node);

        interceptor.before(future, new Object[] { timeout, unit });
        interceptor.after(future, new Object[] { timeout, unit }, null, null);
    } catch (Exception e) {
        fail(e.getMessage());
    }
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
private String getEndPoint(MemcachedNode handlingNode) {
    // TODO duplicated code : ApiInterceptor, FutureGetInterceptor
    final SocketAddress socketAddress = handlingNode.getSocketAddress();
    if (socketAddress instanceof InetSocketAddress) {
        final InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress;
        final String hostAddress = getHostAddress(inetSocketAddress);
        if (hostAddress == null) {
            // TODO return "Unknown Host"; ?
            logger.debug("hostAddress is null");
            return null;
        }
        return HostAndPort.toHostAndPortString(hostAddress, inetSocketAddress.getPort());

    } else {
        if (logger.isDebugEnabled()) {
            logger.debug("invalid socketAddress:{}", socketAddress);
        }
        return null;
    }
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
private String getEndPoint(MemcachedNode handlingNode) {
    // TODO duplicated code : ApiInterceptor, FutureGetInterceptor
    final SocketAddress socketAddress = handlingNode.getSocketAddress();
    if (socketAddress instanceof InetSocketAddress) {
        final InetSocketAddress inetSocketAddress = (InetSocketAddress) socketAddress;
        final String hostAddress = getHostAddress(inetSocketAddress);
        if (hostAddress == null) {
            // TODO return "Unknown Host";
            logger.debug("hostAddress is null");
            return null;
        }
        return HostAndPort.toHostAndPortString(hostAddress, inetSocketAddress.getPort());

    } else {
        if (logger.isDebugEnabled()) {
            logger.debug("invalid socketAddress:{}", socketAddress);
        }
        return null;
    }
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
/**
 * Returns the socket address of a given MemcachedNode.
 *
 * @param node The node which we're interested in
 * @return String the socket address of that node.
 */
protected String getSocketAddressForNode(MemcachedNode node) {
	// Using the internal map retrieve the socket addresses
	// for given nodes.
	// I'm aware that this code is inherently thread-unsafe as
	// I'm using a HashMap implementation of the map, but the worst
	// case ( I believe) is we're slightly in-efficient when
	// a node has never been seen before concurrently on two different
	// threads, so it the socketaddress will be requested multiple times!
	// all other cases should be as fast as possible.
	String result=socketAddresses.get(node);
	if(result == null) {
		result=String.valueOf(node.getSocketAddress());
		result=result.substring(result.indexOf('/') + 1);
		socketAddresses.put(node, result);
	}
	return result;
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
public MemcachedNode getNodeForKey(long _hash) {
    long start = (log.isDebugEnabled()) ? System.nanoTime() : 0;
    try {
        Long hash = Long.valueOf(_hash);
        hash = ketamaNodes.ceilingKey(hash);
        if (hash == null) {
            hash = ketamaNodes.firstKey();
        }
        return ketamaNodes.get(hash);
    } finally {
        if (log.isDebugEnabled()) {
            final long end = System.nanoTime();
            log.debug("getNodeForKey : \t" + (end - start) / 1000);
        }
    }
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
public <T> Future<Boolean> appendOrAdd(String key, CachedData value, int timeToLive, EVCacheLatch evcacheLatch) throws Exception {
    final MemcachedNode node = evcacheMemcachedClient.getEVCacheNode(key);
    if (!ensureWriteQueueSize(node, key)) {
        if (log.isInfoEnabled()) log.info("Node : " + node + " is not active. Failing fast and dropping the write event.");
        final ListenableFuture<Boolean, OperationCompletionListener> defaultFuture = (ListenableFuture<Boolean, OperationCompletionListener>) getDefaultFuture();
        if (evcacheLatch != null && evcacheLatch instanceof EVCacheLatchImpl && !isInWriteOnly()) ((EVCacheLatchImpl) evcacheLatch).addFuture(defaultFuture);
        return defaultFuture;
    }

    try {
        return evcacheMemcachedClient.asyncAppendOrAdd(key, timeToLive, value, evcacheLatch);
    } catch (Exception e) {
        log.error(e.getMessage(), e);
        throw e;
    }
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
private void updateQueueStats() {
    for (ServerGroup serverGroup : memcachedInstancesByServerGroup.keySet()) {
        List<EVCacheClient> clients = memcachedInstancesByServerGroup.get(serverGroup);
        for(EVCacheClient client : clients) {
            final int wSize = client.getWriteQueueLength();
            EVCacheMetricsFactory.getLongGauge("EVCacheClientPool-WriteQueueSize", client.getTagList()).set(Long.valueOf(wSize));
            final int rSize = client.getReadQueueLength();
            EVCacheMetricsFactory.getLongGauge("EVCacheClientPool-ReadQueueSize", client.getTagList()).set(Long.valueOf(rSize));
            if(refreshConnectionOnReadQueueFull.get()) {
                final Collection<MemcachedNode> allNodes = client.getNodeLocator().getAll();
                for (MemcachedNode node : allNodes) {
                    if (node instanceof EVCacheNodeImpl) {
                        final EVCacheNodeImpl evcNode = ((EVCacheNodeImpl) node);
                        if(evcNode.getReadQueueSize() >= refreshConnectionOnReadQueueFullSize.get().intValue()) {
                            EVCacheMetricsFactory.getCounter("EVCacheClientPool-REFRESH_ON_QUEUE_FULL", evcNode.getBaseTags()).increment();
                            client.getEVCacheMemcachedClient().reconnectNode(evcNode);
                        }
                    }
                }
            }
        }
    }
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
private void interruptOldAuth(MemcachedNode nodeToStop) {
  AuthThread toStop = nodeMap.get(nodeToStop);
  if (toStop != null) {
    if (toStop.isAlive()) {
      getLogger().warn(
          "Incomplete authentication interrupted for node " + nodeToStop);
      toStop.interrupt();
    }
    nodeMap.remove(nodeToStop);
  }
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
public AuthThread(MemcachedConnection c, OperationFactory o,
    AuthDescriptor a, MemcachedNode n) {
  conn = c;
  opFact = o;
  authDescriptor = a;
  node = n;
  start();
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
public TapAck(TapConnectionProvider conn, MemcachedNode node,
    TapOpcode opcode, int opaque, OperationCallback cb) {
  this.conn = conn;
  this.node = node;
  this.opcode = opcode;
  this.opaque = opaque;
  this.cb = cb;
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
@Override
public void doInAfterTrace(RecordableTrace trace, Object target, Object[] args, Object result, Throwable throwable) {

    if (parameterExtractor != null) {
        final int index = parameterExtractor.getIndex();
        final Object recordObject = parameterExtractor.extractObject(args);
        trace.recordApi(getMethodDescriptor(), recordObject, index);
    } else {
        trace.recordApi(getMethodDescriptor());
    }

    // find the target node
    if (result instanceof Future && !(result instanceof FrontCacheGetFuture)) {
        final Operation op = getOperation(result);
        if (op != null) {
            MemcachedNode handlingNode = op.getHandlingNode();
            SocketAddress socketAddress = handlingNode.getSocketAddress();
            if (socketAddress instanceof InetSocketAddress) {
                InetSocketAddress address = (InetSocketAddress) socketAddress;
                trace.recordEndPoint(address.getHostName() + ":" + address.getPort());
            }
        } else {
            logger.info("operation not found");
        }
    }

    // determine the service type
    final String serviceCode = getServiceCode(target);
    if (serviceCode != null) {
        trace.recordDestinationId(serviceCode);
        trace.recordServiceType(ServiceType.ARCUS);
    } else {
        trace.recordDestinationId("MEMCACHED");
        trace.recordServiceType(ServiceType.MEMCACHED);
    }

    trace.markAfterTime();
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
private MemcachedNode getMockMemcachedNode() throws IOException {
    java.nio.channels.SocketChannel socketChannel = java.nio.channels.SocketChannel.open();
    BlockingQueue<Operation> readQueue = new LinkedBlockingQueue<Operation>();
    BlockingQueue<Operation> writeQueue = new LinkedBlockingQueue<Operation> ();
    BlockingQueue<Operation> inputQueue = new LinkedBlockingQueue<Operation> ();

    return new AsciiMemcachedNodeImpl(new InetSocketAddress(11211), socketChannel, 128, readQueue, writeQueue, inputQueue, 1000L);
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
@Override
protected void doInAfterTrace(SpanEventRecorder recorder, Object target, Object[] args, Object result, Throwable throwable) {
    recorder.recordApi(methodDescriptor);
    recorder.recordDestinationId("MEMCACHED");
    recorder.recordServiceType(ArcusConstants.MEMCACHED_FUTURE_GET);

    if (!(target instanceof OperationAccessor)) {
        logger.info("operation not found");
        return;
    }

    // find the target node
    final Operation op = ((OperationAccessor) target)._$PINPOINT$_getOperation();
    if (op == null) {
        logger.info("operation is null");
        return;
    }

    recorder.recordException(op.getException());

    final MemcachedNode handlingNode = op.getHandlingNode();
    if (handlingNode != null) {
        final String endPoint = getEndPoint(handlingNode);
        if (endPoint != null) {
            recorder.recordEndPoint(endPoint);
        }
        recorder.recordException(op.getException());
    } else {
        logger.info("no handling node");
    }

    if (op instanceof ServiceCodeAccessor) {
        // determine the service type
        String serviceCode = ((ServiceCodeAccessor) op)._$PINPOINT$_getServiceCode();
        if (serviceCode != null) {
            recorder.recordDestinationId(serviceCode);
            recorder.recordServiceType(ArcusConstants.ARCUS_FUTURE_GET);
        }
    }
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
public int moveOperations(final MemcachedNode toNode) {
	BlockingQueue<Operation> allOp = getAllOperations();
	int opCount = allOp.size();

	if (opCount > 0) {
		toNode.addAllOpToInputQ(allOp);
		getLogger().info("Total %d operations have been moved to %s", opCount, toNode);
	}

	return opCount;
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
private void interruptOldAuth(MemcachedNode nodeToStop) {
	AuthThread toStop = nodeMap.get(nodeToStop);
	if (toStop != null) {
		if (toStop.isAlive()) {
			getLogger().warn("Incomplete authentication interrupted for node " +
				nodeToStop);
			toStop.interrupt();
		}

		nodeMap.remove(nodeToStop);
	}
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
public AuthThread(MemcachedConnection c, OperationFactory o,
		AuthDescriptor a, MemcachedNode n) {
	conn = c;
	opFact = o;
	authDescriptor = a;
	node = n;
	start();
}
 

import net.spy.memcached.MemcachedNode; //导入依赖的package包/类
private static String createMessage(String message,
			Collection<Operation> ops) {
		StringBuilder rv = new StringBuilder(message);
		rv.append(" - failing node");
		rv.append(ops.size() == 1 ? ": " : "s: ");
		boolean first = true;
		for(Operation op : ops) {
			if(first) {
				first = false;
			} else {
				rv.append(", ");
			}
			MemcachedNode node = op == null ? null : op.getHandlingNode();
			rv.append(node == null ? "<unknown>" : node.getSocketAddress());
			if (op != null) {
				rv.append(" [").append(op.getState()).append("]");
			}
			if (node != null) {
				rv.append(" [").append(node.getStatus()).append("]");
			}
//			if (op != null && op.getBuffer() != null) {
//				rv.append(" [")
//						.append(new String(op.getBuffer().array()).replace(
//								"\r\n", "\\n")).append("]");
//			}
		}
		return rv.toString();
	}