import net.spy.memcached.tapmessage.ResponseMessage; //导入依赖的package包/类
/**
 * Gets the next tap message from the queue of received tap messages.
 *
 * @param time the amount of time to wait for a message.
 * @param timeunit the unit of time to use.
 * @return The tap message at the head of the queue or null if the queue is
 *         empty for the given amount of time.
 */
public ResponseMessage getNextMessage(long time, TimeUnit timeunit) {
  try {
    Object m = rqueue.poll(time, timeunit);
    if (m == null) {
      return null;
    } else if (m instanceof ResponseMessage) {
      return (ResponseMessage) m;
    } else if (m instanceof TapAck) {
      TapAck ack = (TapAck) m;
      tapAck(ack.getConn(), ack.getNode(), ack.getOpcode(), ack.getOpaque(),
          ack.getCallback());
      return null;
    } else {
      throw new RuntimeException("Unexpected tap message type");
    }
  } catch (InterruptedException e) {
    shutdown();
    return null;
  }
}
 

import net.spy.memcached.tapmessage.ResponseMessage; //导入依赖的package包/类
/**
 * read messages from CB TAP 
 * @return
 */
List<Event> read() {
	Event e;
	List<Event> result = new ArrayList<Event>(1);
	if(tapClient.hasMoreMessages()){
		//System.out.println("tap client has messages..");
		ResponseMessage resmessage=tapClient.getNextMessage();
		
		//System.out.println("[message]"+resmessage);
		if(resmessage!=null ) {
			e=CBMessageConverter.convert(filter,resmessage);
			if(e!=null)result.add(e);
		}
	}
	return result;
}
 

import net.spy.memcached.tapmessage.ResponseMessage; //导入依赖的package包/类
public static Event convert(CBMessageFilter filter, ResponseMessage message) {
	//System.out.println("[message]"+message.getKey());
	Event event = new SimpleEvent();
	Map<String, String> headers = event.getHeaders();
	
	String key=message.getKey();
	key=key.substring(0,key.indexOf("_",key.indexOf("_")+1));
	//System.out.println("Key is "+key);
	
	if(filter.membershiptest(key)){
		String body="{ \"key\"={"+message.getKey()+"},\"value\"={"+ new String(message.getValue())+"}}";
		event.setBody(body.getBytes());
		return event;
	}else{
		return null;
	}
	
	/*
	//Logic specific to cookie . Generalize it.
	if(message.getKey().startsWith("cs_ca_")){
		headers.put("type", "analytics");
	}*/
	
	
}
 

import net.spy.memcached.tapmessage.ResponseMessage; //导入依赖的package包/类
@Override
public void readFromBuffer(ByteBuffer data) throws IOException {
  while (data.remaining() > 0) {
    if (bytesProcessed < BaseMessage.HEADER_LENGTH) {
      header[bytesProcessed] = data.get();
      bytesProcessed++;
    } else {
      if (message == null) {
        bodylen = decodeInt(header, 8);
        message = new byte[BaseMessage.HEADER_LENGTH + bodylen];
        System.arraycopy(header, 0, message, 0, BaseMessage.HEADER_LENGTH);
      }

      if (bytesProcessed < message.length) {
        message[bytesProcessed] = data.get();
        bytesProcessed++;
      }
      if (bytesProcessed >= message.length) {
        ResponseMessage response = new ResponseMessage(message);
        for (TapResponseFlag flag : response.getFlags()) {
          if (flag == TapResponseFlag.TAP_ACK) {
            ((Callback) getCallback()).gotAck(getHandlingNode(),
                response.getOpcode(), response.getOpaque());
          }
        }
        if (response.getOpcode() != TapOpcode.OPAQUE && response.getOpcode()
          != TapOpcode.NOOP) {
          ((Callback) getCallback()).gotData(response);
        }
        message = null;
        bytesProcessed = 0;
      }
    }
  }
}
 

import net.spy.memcached.tapmessage.ResponseMessage; //导入依赖的package包/类
/**
 * Allows the user to specify a custom tap message.
 *
 * @param id the named tap id that can be used to resume a disconnected tap
 *          stream
 * @param message the custom tap message that will be used to initiate the tap
 *          stream.
 * @return the operation that controls the tap stream.
 * @throws ConfigurationException a bad configuration was received from the
 *           memcached cluster.
 * @throws IOException if there are errors connecting to the cluster.
 */
public TapStream tapCustom(final String id, final RequestMessage message)
  throws ConfigurationException, IOException {
  final TapConnectionProvider conn = new TapConnectionProvider(addrs);
  final TapStream ts = new TapStream();
  conn.broadcastOp(new BroadcastOpFactory() {
    public Operation newOp(final MemcachedNode n,
        final CountDownLatch latch) {
      Operation op =  conn.getOpFactory().tapCustom(id, message,
          new TapOperation.Callback() {
          public void receivedStatus(OperationStatus status) {
          }
          public void gotData(ResponseMessage tapMessage) {
            rqueue.add(tapMessage);
            messagesRead++;
          }
          public void gotAck(MemcachedNode node, TapOpcode opcode,
              int opaque) {
            rqueue.add(new TapAck(conn, node, opcode, opaque, this));
          }
          public void complete() {
            latch.countDown();
          }
        });
      ts.addOp((TapOperation)op);
      return op;
    }
  });
  synchronized (omap) {
    omap.put(ts, conn);
  }
  return ts;
}
 

import net.spy.memcached.tapmessage.ResponseMessage; //导入依赖的package包/类
/**
 * Specifies a tap stream that will take a snapshot of items in memcached and
 * send them through a tap stream.
 *
 * @param id the named tap id that can be used to resume a disconnected tap
 *          stream
 * @return the operation that controls the tap stream.
 * @throws ConfigurationException a bad configuration was received from the
 *           memcached cluster.
 * @throws IOException If there are errors connecting to the cluster.
 */
public TapStream tapDump(final String id) throws IOException,
    ConfigurationException {
  final TapConnectionProvider conn = new TapConnectionProvider(addrs);
  final TapStream ts = new TapStream();
  conn.broadcastOp(new BroadcastOpFactory() {
    public Operation newOp(final MemcachedNode n,
        final CountDownLatch latch) {
      Operation op =  conn.getOpFactory().tapDump(id,
          new TapOperation.Callback() {
          public void receivedStatus(OperationStatus status) {
          }
          public void gotData(ResponseMessage tapMessage) {
            rqueue.add(tapMessage);
            messagesRead++;
          }
          public void gotAck(MemcachedNode node, TapOpcode opcode,
              int opaque) {
            rqueue.add(new TapAck(conn, node, opcode, opaque, this));
          }
          public void complete() {
            latch.countDown();
          }
        });
      ts.addOp((TapOperation)op);
      return op;
    }
  });
  synchronized (omap) {
    omap.put(ts, conn);
  }
  return ts;
}
 

import net.spy.memcached.tapmessage.ResponseMessage; //导入依赖的package包/类
@RequestMapping(value = "/{table}/add", method = RequestMethod.GET)
public ResponseMessage add(){
    return null;
}
 

import net.spy.memcached.tapmessage.ResponseMessage; //导入依赖的package包/类
@Override
public int next() {
  Stopwatch watch = new Stopwatch();
  watch.start();

  keyVector.clear();
  keyVector.allocateNew();
  valueVector.clear();
  valueVector.allocateNew();
  int rowCount = 0;
  done:
  for (; rowCount < TARGET_RECORD_COUNT && tapClient.hasMoreMessages();) {
    ResponseMessage message = null;
    if (leftOver != null) {
      message = leftOver;
      leftOver = null;
    } else {
      if ((message = tapClient.getNextMessage()) == null) {
        continue;
      }
    }

    if (!keyVector.getMutator().setSafe(rowCount, message.getKey().getBytes())) {
      setOutputRowCount(rowCount);
      leftOver = message;
      break done;
    }

    if (!valueVector.getMutator().setSafe(rowCount, message.getValue())) {
      setOutputRowCount(rowCount);
      leftOver = message;
      break done;
    }

    rowCount++;
  }

  setOutputRowCount(rowCount);
  logger.debug("Took {} ms to get {} records", watch.elapsed(TimeUnit.MILLISECONDS), rowCount);
  return rowCount;
}
 

import net.spy.memcached.tapmessage.ResponseMessage; //导入依赖的package包/类
/**
 * Callback for each result from a get.
 *
 * @param message the response message sent from the server
 */
void gotData(ResponseMessage message);