geode-core/src/main/java/org/apache/geode/internal/cache/tx/RemoteContainsKeyValueMessage.java - geode - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
  * agreements. See the NOTICE file distributed with this work for additional information regarding
  * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance with the License. You may obtain a
  * copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software distributed under the License
  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  * or implied. See the License for the specific language governing permissions and limitations under
  * the License.
  */
 package org.apache.geode.internal.cache.tx;

 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Set;

 import org.apache.logging.log4j.Logger;

 import org.apache.geode.DataSerializer;
 import org.apache.geode.cache.CacheException;
 import org.apache.geode.distributed.internal.ClusterDistributionManager;
 import org.apache.geode.distributed.internal.DirectReplyProcessor;
 import org.apache.geode.distributed.internal.DistributionManager;
 import org.apache.geode.distributed.internal.DistributionMessage;
 import org.apache.geode.distributed.internal.DistributionStats;
 import org.apache.geode.distributed.internal.InternalDistributedSystem;
 import org.apache.geode.distributed.internal.ReplyMessage;
 import org.apache.geode.distributed.internal.ReplyProcessor21;
 import org.apache.geode.distributed.internal.ReplySender;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
 import org.apache.geode.internal.Assert;
 import org.apache.geode.internal.cache.LocalRegion;
 import org.apache.geode.internal.cache.PartitionedRegion;
 import org.apache.geode.internal.cache.RemoteOperationException;
 import org.apache.geode.internal.logging.LogService;
 import org.apache.geode.internal.logging.log4j.LogMarker;

 /**
  * This message is used be a replicate region to send a contains key/value request to another peer.
  *
  * @since GemFire 6.5
  */
 public class RemoteContainsKeyValueMessage extends RemoteOperationMessageWithDirectReply {
   private static final Logger logger = LogService.getLogger();

   private boolean valueCheck;

   private Object key;

   protected static final short VALUE_CHECK = UNRESERVED_FLAGS_START;

   public RemoteContainsKeyValueMessage() {}

   public RemoteContainsKeyValueMessage(InternalDistributedMember recipient, String regionPath,
       DirectReplyProcessor processor, Object key, boolean valueCheck) {
     super(recipient, regionPath, processor);
     this.valueCheck = valueCheck;
     this.key = key;
   }

   /**
    * Sends a ReplicateRegion message for either
    * {@link org.apache.geode.cache.Region#containsKey(Object)}or
    * {@link org.apache.geode.cache.Region#containsValueForKey(Object)} depending on the
    * <code>valueCheck</code> argument
    *
    * @param recipient the member that the contains keys/value message is sent to
    * @param r the LocalRegion
    * @param key the key to be queried
    * @param valueCheck true if {@link org.apache.geode.cache.Region#containsValueForKey(Object)} is
    *        desired, false if {@link org.apache.geode.cache.Region#containsKey(Object)}is desired
    * @return the processor used to read the returned keys
    */
   public static RemoteContainsKeyValueResponse send(InternalDistributedMember recipient,
       LocalRegion r, Object key, boolean valueCheck) throws RemoteOperationException {
     Assert.assertTrue(recipient != null, "recipient can not be NULL");

     RemoteContainsKeyValueResponse p =
         new RemoteContainsKeyValueResponse(r.getSystem(), recipient, key);
     RemoteContainsKeyValueMessage m =
         new RemoteContainsKeyValueMessage(recipient, r.getFullPath(), p, key, valueCheck);

     Set<?> failures = r.getDistributionManager().putOutgoing(m);
     if (failures != null && failures.size() > 0) {
       throw new RemoteOperationException(String.format("Failed sending < %s >", m));
     }
     return p;
   }

   @Override
   protected boolean operateOnRegion(ClusterDistributionManager dm, LocalRegion r, long startTime)
       throws CacheException, RemoteOperationException {
     if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
       logger.trace(LogMarker.DM_VERBOSE,
           "DistributedRemoteContainsKeyValueMessage operateOnRegion: {}", r.getFullPath());
     }

     if (!(r instanceof PartitionedRegion)) { // prs already wait on initialization
       r.waitOnInitialization(); // bug #43371 - accessing a region before it's initialized
     }

     final boolean replyVal;
     if (this.valueCheck) {
       replyVal = r.containsValueForKey(this.key);
     } else {
       replyVal = r.containsKey(this.key);
     }

     if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
       logger.trace(
           "DistributedRemoteContainsKeyValueMessage sending reply back using processorId: {}",
           getProcessorId());
     }

     RemoteContainsKeyValueReplyMessage.send(getSender(), getProcessorId(), getReplySender(dm),
         replyVal);

     // Unless there was an exception thrown, this message handles sending the
     // response
     return false;
   }

   @Override
   protected void appendFields(StringBuffer buff) {
     super.appendFields(buff);
     buff.append("; valueCheck=").append(this.valueCheck).append("; key=").append(this.key);
   }

   @Override
   public int getDSFID() {
     return R_CONTAINS_MESSAGE;
   }

   @Override
   public void fromData(DataInput in) throws IOException, ClassNotFoundException {
     super.fromData(in);
     this.key = DataSerializer.readObject(in);
     this.valueCheck = (flags & VALUE_CHECK) != 0;
   }

   @Override
   public void toData(DataOutput out) throws IOException {
     super.toData(out);
     DataSerializer.writeObject(this.key, out);
   }

   @Override
   protected short computeCompressedShort() {
     short flags = super.computeCompressedShort();
     if (this.valueCheck)
       flags |= VALUE_CHECK;
     return flags;
   }

   public static class RemoteContainsKeyValueReplyMessage extends ReplyMessage {

     /** Propagated exception from remote node to operation initiator */
     private boolean containsKeyValue;

     /**
      * Empty constructor to conform to DataSerializable interface
      */
     public RemoteContainsKeyValueReplyMessage() {}

     private RemoteContainsKeyValueReplyMessage(int processorId, boolean containsKeyValue) {
       this.processorId = processorId;
       this.containsKeyValue = containsKeyValue;
     }

     /** Send an ack */
     public static void send(InternalDistributedMember recipient, int processorId,
         ReplySender replySender, boolean containsKeyValue) {
       Assert.assertTrue(recipient != null, "ContainsKeyValueReplyMessage NULL reply message");
       RemoteContainsKeyValueReplyMessage m =
           new RemoteContainsKeyValueReplyMessage(processorId, containsKeyValue);
       m.setRecipient(recipient);
       replySender.putOutgoing(m);
     }

     /**
      * Processes this message. This method is invoked by the receiver of the message.
      *
      * @param dm the distribution manager that is processing the message.
      */
     @Override
     public void process(final DistributionManager dm, ReplyProcessor21 processor) {
       final long startTime = getTimestamp();

       if (processor == null) {
         if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
           logger.trace(LogMarker.DM_VERBOSE, "ContainsKeyValueReplyMessage processor not found");
         }
         return;
       }
       processor.process(this);

       dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
     }

     @Override
     public int getDSFID() {
       return R_CONTAINS_REPLY_MESSAGE;
     }

     @Override
     public void fromData(DataInput in) throws IOException, ClassNotFoundException {
       super.fromData(in);
       this.containsKeyValue = in.readBoolean();
     }

     @Override
     public void toData(DataOutput out) throws IOException {
       super.toData(out);
       out.writeBoolean(this.containsKeyValue);
     }

     @Override
     public String toString() {
       StringBuffer sb = new StringBuffer();
       sb.append("ContainsKeyValueReplyMessage ").append("processorid=").append(this.processorId)
           .append(" reply to sender ").append(this.getSender())
           .append(" returning containsKeyValue=").append(doesItContainKeyValue());
       return sb.toString();
     }

     public boolean doesItContainKeyValue() {
       return this.containsKeyValue;
     }
   }

   /**
    * A processor to capture the value returned by
    * {@link org.apache.geode.internal.cache.tx.RemoteContainsKeyValueMessage.RemoteContainsKeyValueReplyMessage}
    *
    * @since GemFire 6.5
    */
   public static class RemoteContainsKeyValueResponse extends RemoteOperationResponse {
     private volatile boolean returnValue;
     private volatile boolean returnValueReceived;
     final Object key;

     public RemoteContainsKeyValueResponse(InternalDistributedSystem ds,
         InternalDistributedMember recipient, Object key) {
       super(ds, recipient, false);
       this.key = key;
     }

     @Override
     public void process(DistributionMessage msg) {
       try {
         if (msg instanceof RemoteContainsKeyValueReplyMessage) {
           RemoteContainsKeyValueReplyMessage reply = (RemoteContainsKeyValueReplyMessage) msg;
           this.returnValue = reply.doesItContainKeyValue();
           this.returnValueReceived = true;
           if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
             logger.trace(LogMarker.DM_VERBOSE, "ContainsKeyValueResponse return value is {}",
                 this.returnValue);
           }
         }
       } finally {
         super.process(msg);
       }
     }

     /**
      * @return Set the keys associated with the ReplicateRegion of the
      *         {@link RemoteContainsKeyValueMessage}
      */
     public boolean waitForContainsResult() throws RemoteOperationException {
       try {
         waitForRemoteResponse();
       } catch (CacheException ce) {
         logger.debug("ContainsKeyValueResponse got remote CacheException", ce);
         throw new RemoteOperationException(
             "RemoteContainsKeyResponse got remote CacheException; triggering RemoteOperationException.",
             ce);
       }
       if (!this.returnValueReceived) {
         throw new RemoteOperationException(
             "no return value received");
       }
       return this.returnValue;
     }
   }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more contributor license
	* agreements. See the NOTICE file distributed with this work for additional information regarding
	* copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance with the License. You may obtain a
	* copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software distributed under the License
	* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
	* or implied. See the License for the specific language governing permissions and limitations under
	* the License.
	*/
	package org.apache.geode.internal.cache.tx;

	import java.io.DataInput;
	import java.io.DataOutput;
	import java.io.IOException;
	import java.util.Set;

	import org.apache.logging.log4j.Logger;

	import org.apache.geode.DataSerializer;
	import org.apache.geode.cache.CacheException;
	import org.apache.geode.distributed.internal.ClusterDistributionManager;
	import org.apache.geode.distributed.internal.DirectReplyProcessor;
	import org.apache.geode.distributed.internal.DistributionManager;
	import org.apache.geode.distributed.internal.DistributionMessage;
	import org.apache.geode.distributed.internal.DistributionStats;
	import org.apache.geode.distributed.internal.InternalDistributedSystem;
	import org.apache.geode.distributed.internal.ReplyMessage;
	import org.apache.geode.distributed.internal.ReplyProcessor21;
	import org.apache.geode.distributed.internal.ReplySender;
	import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
	import org.apache.geode.internal.Assert;
	import org.apache.geode.internal.cache.LocalRegion;
	import org.apache.geode.internal.cache.PartitionedRegion;
	import org.apache.geode.internal.cache.RemoteOperationException;
	import org.apache.geode.internal.logging.LogService;
	import org.apache.geode.internal.logging.log4j.LogMarker;

	/**
	* This message is used be a replicate region to send a contains key/value request to another peer.
	*
	* @since GemFire 6.5
	*/
	public class RemoteContainsKeyValueMessage extends RemoteOperationMessageWithDirectReply {
	private static final Logger logger = LogService.getLogger();

	private boolean valueCheck;

	private Object key;

	protected static final short VALUE_CHECK = UNRESERVED_FLAGS_START;

	public RemoteContainsKeyValueMessage() {}

	public RemoteContainsKeyValueMessage(InternalDistributedMember recipient, String regionPath,
	DirectReplyProcessor processor, Object key, boolean valueCheck) {
	super(recipient, regionPath, processor);
	this.valueCheck = valueCheck;
	this.key = key;
	}

	/**
	* Sends a ReplicateRegion message for either
	* {@link org.apache.geode.cache.Region#containsKey(Object)}or
	* {@link org.apache.geode.cache.Region#containsValueForKey(Object)} depending on the
	* <code>valueCheck</code> argument
	*
	* @param recipient the member that the contains keys/value message is sent to
	* @param r the LocalRegion
	* @param key the key to be queried
	* @param valueCheck true if {@link org.apache.geode.cache.Region#containsValueForKey(Object)} is
	* desired, false if {@link org.apache.geode.cache.Region#containsKey(Object)}is desired
	* @return the processor used to read the returned keys
	*/
	public static RemoteContainsKeyValueResponse send(InternalDistributedMember recipient,
	LocalRegion r, Object key, boolean valueCheck) throws RemoteOperationException {
	Assert.assertTrue(recipient != null, "recipient can not be NULL");

	RemoteContainsKeyValueResponse p =
	new RemoteContainsKeyValueResponse(r.getSystem(), recipient, key);
	RemoteContainsKeyValueMessage m =
	new RemoteContainsKeyValueMessage(recipient, r.getFullPath(), p, key, valueCheck);

	Set<?> failures = r.getDistributionManager().putOutgoing(m);
	if (failures != null && failures.size() > 0) {
	throw new RemoteOperationException(String.format("Failed sending < %s >", m));
	}
	return p;
	}

	@Override
	protected boolean operateOnRegion(ClusterDistributionManager dm, LocalRegion r, long startTime)
	throws CacheException, RemoteOperationException {
	if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
	logger.trace(LogMarker.DM_VERBOSE,
	"DistributedRemoteContainsKeyValueMessage operateOnRegion: {}", r.getFullPath());
	}

	if (!(r instanceof PartitionedRegion)) { // prs already wait on initialization
	r.waitOnInitialization(); // bug #43371 - accessing a region before it's initialized
	}

	final boolean replyVal;
	if (this.valueCheck) {
	replyVal = r.containsValueForKey(this.key);
	} else {
	replyVal = r.containsKey(this.key);
	}

	if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
	logger.trace(
	"DistributedRemoteContainsKeyValueMessage sending reply back using processorId: {}",
	getProcessorId());
	}

	RemoteContainsKeyValueReplyMessage.send(getSender(), getProcessorId(), getReplySender(dm),
	replyVal);

	// Unless there was an exception thrown, this message handles sending the
	// response
	return false;
	}

	@Override
	protected void appendFields(StringBuffer buff) {
	super.appendFields(buff);
	buff.append("; valueCheck=").append(this.valueCheck).append("; key=").append(this.key);
	}

	@Override
	public int getDSFID() {
	return R_CONTAINS_MESSAGE;
	}

	@Override
	public void fromData(DataInput in) throws IOException, ClassNotFoundException {
	super.fromData(in);
	this.key = DataSerializer.readObject(in);
	this.valueCheck = (flags & VALUE_CHECK) != 0;
	}

	@Override
	public void toData(DataOutput out) throws IOException {
	super.toData(out);
	DataSerializer.writeObject(this.key, out);
	}

	@Override
	protected short computeCompressedShort() {
	short flags = super.computeCompressedShort();
	if (this.valueCheck)
	flags \|= VALUE_CHECK;
	return flags;
	}

	public static class RemoteContainsKeyValueReplyMessage extends ReplyMessage {

	/** Propagated exception from remote node to operation initiator */
	private boolean containsKeyValue;

	/**
	* Empty constructor to conform to DataSerializable interface
	*/
	public RemoteContainsKeyValueReplyMessage() {}

	private RemoteContainsKeyValueReplyMessage(int processorId, boolean containsKeyValue) {
	this.processorId = processorId;
	this.containsKeyValue = containsKeyValue;
	}

	/** Send an ack */
	public static void send(InternalDistributedMember recipient, int processorId,
	ReplySender replySender, boolean containsKeyValue) {
	Assert.assertTrue(recipient != null, "ContainsKeyValueReplyMessage NULL reply message");
	RemoteContainsKeyValueReplyMessage m =
	new RemoteContainsKeyValueReplyMessage(processorId, containsKeyValue);
	m.setRecipient(recipient);
	replySender.putOutgoing(m);
	}

	/**
	* Processes this message. This method is invoked by the receiver of the message.
	*
	* @param dm the distribution manager that is processing the message.
	*/
	@Override
	public void process(final DistributionManager dm, ReplyProcessor21 processor) {
	final long startTime = getTimestamp();

	if (processor == null) {
	if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
	logger.trace(LogMarker.DM_VERBOSE, "ContainsKeyValueReplyMessage processor not found");
	}
	return;
	}
	processor.process(this);

	dm.getStats().incReplyMessageTime(DistributionStats.getStatTime() - startTime);
	}

	@Override
	public int getDSFID() {
	return R_CONTAINS_REPLY_MESSAGE;
	}

	@Override
	public void fromData(DataInput in) throws IOException, ClassNotFoundException {
	super.fromData(in);
	this.containsKeyValue = in.readBoolean();
	}

	@Override
	public void toData(DataOutput out) throws IOException {
	super.toData(out);
	out.writeBoolean(this.containsKeyValue);
	}

	@Override
	public String toString() {
	StringBuffer sb = new StringBuffer();
	sb.append("ContainsKeyValueReplyMessage ").append("processorid=").append(this.processorId)
	.append(" reply to sender ").append(this.getSender())
	.append(" returning containsKeyValue=").append(doesItContainKeyValue());
	return sb.toString();
	}

	public boolean doesItContainKeyValue() {
	return this.containsKeyValue;
	}
	}

	/**
	* A processor to capture the value returned by
	* {@link org.apache.geode.internal.cache.tx.RemoteContainsKeyValueMessage.RemoteContainsKeyValueReplyMessage}
	*
	* @since GemFire 6.5
	*/
	public static class RemoteContainsKeyValueResponse extends RemoteOperationResponse {
	private volatile boolean returnValue;
	private volatile boolean returnValueReceived;
	final Object key;

	public RemoteContainsKeyValueResponse(InternalDistributedSystem ds,
	InternalDistributedMember recipient, Object key) {
	super(ds, recipient, false);
	this.key = key;
	}

	@Override
	public void process(DistributionMessage msg) {
	try {
	if (msg instanceof RemoteContainsKeyValueReplyMessage) {
	RemoteContainsKeyValueReplyMessage reply = (RemoteContainsKeyValueReplyMessage) msg;
	this.returnValue = reply.doesItContainKeyValue();
	this.returnValueReceived = true;
	if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
	logger.trace(LogMarker.DM_VERBOSE, "ContainsKeyValueResponse return value is {}",
	this.returnValue);
	}
	}
	} finally {
	super.process(msg);
	}
	}

	/**
	* @return Set the keys associated with the ReplicateRegion of the
	* {@link RemoteContainsKeyValueMessage}
	*/
	public boolean waitForContainsResult() throws RemoteOperationException {
	try {
	waitForRemoteResponse();
	} catch (CacheException ce) {
	logger.debug("ContainsKeyValueResponse got remote CacheException", ce);
	throw new RemoteOperationException(
	"RemoteContainsKeyResponse got remote CacheException; triggering RemoteOperationException.",
	ce);
	}
	if (!this.returnValueReceived) {
	throw new RemoteOperationException(
	"no return value received");
	}
	return this.returnValue;
	}
	}

	}