code: 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;
    }
docs: 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.