| /* |
| * 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 static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.ENTRY_EVENT_OLD_VALUE; |
| |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Set; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.cache.CacheException; |
| import org.apache.geode.cache.CacheWriterException; |
| import org.apache.geode.cache.EntryExistsException; |
| import org.apache.geode.cache.EntryNotFoundException; |
| import org.apache.geode.cache.Operation; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.TransactionDataNotColocatedException; |
| import org.apache.geode.distributed.DistributedMember; |
| 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.InternalDistributedSystem; |
| import org.apache.geode.distributed.internal.ReplyException; |
| 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.InternalDataSerializer; |
| import org.apache.geode.internal.NanoTimer; |
| import org.apache.geode.internal.cache.AbstractRegion; |
| import org.apache.geode.internal.cache.CachedDeserializable; |
| import org.apache.geode.internal.cache.DataLocationException; |
| import org.apache.geode.internal.cache.DistributedCacheOperation; |
| import org.apache.geode.internal.cache.DistributedRegion; |
| import org.apache.geode.internal.cache.EntryEventImpl; |
| import org.apache.geode.internal.cache.EntryEventImpl.OldValueImporter; |
| import org.apache.geode.internal.cache.EventID; |
| import org.apache.geode.internal.cache.InternalRegion; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.RemoteOperationException; |
| import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID; |
| import org.apache.geode.internal.cache.versions.DiskVersionTag; |
| import org.apache.geode.internal.cache.versions.VersionTag; |
| import org.apache.geode.internal.logging.log4j.LogMarker; |
| import org.apache.geode.internal.offheap.annotations.Released; |
| import org.apache.geode.internal.offheap.annotations.Unretained; |
| import org.apache.geode.internal.serialization.DeserializationContext; |
| import org.apache.geode.internal.serialization.SerializationContext; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| |
| /** |
| * A class that specifies a destroy operation. Used by ReplicateRegions. Note: The reason for |
| * different classes for Destroy and Invalidate is to prevent sending an extra bit for every |
| * RemoteDestroyMessage to differentiate an invalidate versus a destroy. The assumption is that |
| * these operations are used frequently, if they are not then it makes sense to fold the destroy and |
| * the invalidate into the same message and use an extra bit to differentiate. |
| * |
| * This message is used by transactions to destroy an entry on a transaction hosted on a remote |
| * member. It is also used by non-transactional region destroys that need to generate a VersionTag |
| * on a remote member. |
| * |
| * @since GemFire 6.5 |
| * |
| */ |
| public class RemoteDestroyMessage extends RemoteOperationMessageWithDirectReply |
| implements OldValueImporter { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| /** The key associated with the value that must be sent */ |
| private Object key; |
| |
| /** The callback arg */ |
| private Object cbArg; |
| |
| /** The operation performed on the sender */ |
| private Operation op; |
| |
| /** |
| * An additional object providing context for the operation, e.g., for BridgeServer notification |
| */ |
| ClientProxyMembershipID bridgeContext; |
| |
| /** event identifier */ |
| EventID eventId; |
| |
| /** for relayed messages, this is the original sender of the message */ |
| InternalDistributedMember originalSender; |
| |
| /** whether the message has old value */ |
| private boolean hasOldValue = false; |
| |
| /** whether old value is serialized */ |
| private boolean oldValueIsSerialized = false; |
| |
| private Object expectedOldValue; |
| |
| private byte[] oldValBytes; |
| |
| @Unretained(ENTRY_EVENT_OLD_VALUE) |
| private transient Object oldValObj; |
| |
| boolean useOriginRemote; |
| |
| protected boolean possibleDuplicate; |
| |
| VersionTag<?> versionTag; |
| |
| // additional bitmask flags used for serialization/deserialization |
| |
| protected static final short USE_ORIGIN_REMOTE = UNRESERVED_FLAGS_START; |
| protected static final short HAS_OLD_VALUE = (USE_ORIGIN_REMOTE << 1); |
| protected static final short CACHE_WRITE = (HAS_OLD_VALUE << 1); |
| protected static final short HAS_BRIDGE_CONTEXT = (CACHE_WRITE << 1); |
| protected static final short HAS_ORIGINAL_SENDER = (HAS_BRIDGE_CONTEXT << 1); |
| protected static final int HAS_VERSION_TAG = (HAS_ORIGINAL_SENDER << 1); |
| |
| /** |
| * Empty constructor to satisfy {@link DataSerializer} requirements |
| */ |
| public RemoteDestroyMessage() {} |
| |
| protected RemoteDestroyMessage(DistributedMember recipient, String regionPath, |
| DirectReplyProcessor processor, EntryEventImpl event, Object expectedOldValue, |
| boolean useOriginRemote, boolean possibleDuplicate) { |
| super((InternalDistributedMember) recipient, regionPath, processor); |
| this.expectedOldValue = expectedOldValue; |
| this.key = event.getKey(); |
| this.cbArg = event.getRawCallbackArgument(); |
| this.op = event.getOperation(); |
| this.bridgeContext = event.getContext(); |
| this.eventId = event.getEventId(); |
| this.useOriginRemote = useOriginRemote; |
| this.possibleDuplicate = possibleDuplicate; |
| this.versionTag = event.getVersionTag(); |
| Assert.assertTrue(this.eventId != null); |
| |
| // added for old value if available sent over the wire for cache servers. |
| if (event.hasOldValue()) { |
| this.hasOldValue = true; |
| event.exportOldValue(this); |
| } |
| } |
| |
| private void setOldValBytes(byte[] valBytes) { |
| this.oldValBytes = valBytes; |
| } |
| |
| private void setOldValObj(@Unretained(ENTRY_EVENT_OLD_VALUE) Object o) { |
| this.oldValObj = o; |
| } |
| |
| public byte[] getOldValueBytes() { |
| return this.oldValBytes; |
| } |
| |
| private Object getOldValObj() { |
| return this.oldValObj; |
| } |
| |
| protected boolean getHasOldValue() { |
| return this.hasOldValue; |
| } |
| |
| protected boolean getOldValueIsSerialized() { |
| return this.oldValueIsSerialized; |
| } |
| |
| /** |
| * Set the old value for this message, only used if there are cqs registered on one of the bridge |
| * servers. |
| * |
| * @param event underlying event. |
| * @since GemFire 5.5 |
| */ |
| public void setOldValue(EntryEventImpl event) { |
| if (event.hasOldValue()) { |
| this.hasOldValue = true; |
| CachedDeserializable cd = (CachedDeserializable) event.getSerializedOldValue(); |
| if (cd != null) { |
| if (!cd.isSerialized()) { |
| // it is a byte[] |
| this.oldValueIsSerialized = false; |
| setOldValBytes((byte[]) cd.getDeserializedForReading()); |
| } else { |
| this.oldValueIsSerialized = true; |
| Object o = cd.getValue(); |
| if (o instanceof byte[]) { |
| setOldValBytes((byte[]) o); |
| } else { |
| // Defer serialization until toData is called. |
| setOldValObj(o); |
| } |
| } |
| } else { |
| Object old = event.getRawOldValue(); |
| if (old instanceof byte[]) { |
| this.oldValueIsSerialized = false; |
| setOldValBytes((byte[]) old); |
| } else { |
| this.oldValueIsSerialized = true; |
| setOldValObj(AbstractRegion.handleNotAvailable(old)); |
| } |
| } |
| } |
| } |
| |
| |
| @SuppressWarnings("unchecked") |
| public static boolean distribute(EntryEventImpl event, Object expectedOldValue, |
| boolean onlyPersistent) { |
| boolean successful = false; |
| DistributedRegion r = (DistributedRegion) event.getRegion(); |
| Collection<InternalDistributedMember> replicates = onlyPersistent |
| ? r.getCacheDistributionAdvisor().adviseInitializedPersistentMembers().keySet() |
| : r.getCacheDistributionAdvisor().adviseInitializedReplicates(); |
| if (replicates.isEmpty()) { |
| return false; |
| } |
| if (replicates.size() > 1) { |
| ArrayList<InternalDistributedMember> l = new ArrayList<>(replicates); |
| Collections.shuffle(l); |
| replicates = l; |
| } |
| int attempts = 0; |
| for (InternalDistributedMember replicate : replicates) { |
| try { |
| attempts++; |
| final boolean posDup = (attempts > 1); |
| RemoteDestroyReplyProcessor processor = |
| send(replicate, event.getRegion(), event, expectedOldValue, false, posDup); |
| processor.waitForRemoteResponse(); |
| VersionTag<?> versionTag = processor.getVersionTag(); |
| if (versionTag != null) { |
| event.setVersionTag(versionTag); |
| if (event.getRegion().getVersionVector() != null) { |
| event.getRegion().getVersionVector().recordVersion(versionTag.getMemberID(), |
| versionTag); |
| } |
| } |
| event.setInhibitDistribution(true); |
| return true; |
| |
| } catch (EntryNotFoundException e) { |
| throw new EntryNotFoundException("" + event.getKey()); |
| |
| } catch (TransactionDataNotColocatedException enfe) { |
| throw enfe; |
| |
| } catch (CancelException e) { |
| event.getRegion().getCancelCriterion().checkCancelInProgress(e); |
| |
| } catch (CacheException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("RemoteDestroyMessage caught CacheException during distribution", e); |
| } |
| successful = true; // not a cancel-exception, so don't complain any more about it |
| |
| } catch (RegionDestroyedException | RemoteOperationException e) { |
| if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) { |
| logger.trace(LogMarker.DM_VERBOSE, |
| "RemoteDestroyMessage caught an exception during distribution; retrying to another member", |
| e); |
| } |
| } |
| } |
| return successful; |
| } |
| |
| /** |
| * Sends a RemoteDestroyMessage {@link org.apache.geode.cache.Region#destroy(Object)}message to |
| * the recipient |
| * |
| * @param recipient the recipient of the message |
| * @param r the ReplicateRegion for which the destroy was performed |
| * @param event the event causing this message |
| * @return the processor used to await the potential {@link org.apache.geode.cache.CacheException} |
| */ |
| public static RemoteDestroyReplyProcessor send(DistributedMember recipient, InternalRegion r, |
| EntryEventImpl event, Object expectedOldValue, boolean useOriginRemote, |
| boolean possibleDuplicate) throws RemoteOperationException { |
| RemoteDestroyReplyProcessor p = |
| new RemoteDestroyReplyProcessor(r.getSystem(), recipient, false); |
| p.requireResponse(); |
| RemoteDestroyMessage m = new RemoteDestroyMessage(recipient, r.getFullPath(), p, event, |
| expectedOldValue, useOriginRemote, possibleDuplicate); |
| Set<?> failures = r.getDistributionManager().putOutgoing(m); |
| if (failures != null && failures.size() > 0) { |
| throw new RemoteOperationException( |
| String.format("Failed sending < %s >", m)); |
| } |
| return p; |
| } |
| |
| /** |
| * This method is called upon receipt and make the desired changes to the region. Note: It is very |
| * important that this message does NOT cause any deadlocks as the sender will wait indefinitely |
| * for the acknowledgement |
| */ |
| @Override |
| protected boolean operateOnRegion(ClusterDistributionManager dm, LocalRegion r, long startTime) |
| throws EntryExistsException, RemoteOperationException { |
| InternalDistributedMember eventSender = originalSender; |
| if (eventSender == null) { |
| eventSender = getSender(); |
| } |
| @Released |
| EntryEventImpl event = null; |
| try { |
| if (this.bridgeContext != null) { |
| event = EntryEventImpl.create(r, getOperation(), getKey(), null/* newValue */, |
| getCallbackArg(), false/* originRemote */, eventSender, true/* generateCallbacks */); |
| event.setContext(this.bridgeContext); |
| |
| // for cq processing and client notification by BS. |
| if (this.hasOldValue) { |
| if (this.oldValueIsSerialized) { |
| event.setSerializedOldValue(getOldValueBytes()); |
| } else { |
| event.setOldValue(getOldValueBytes()); |
| } |
| } |
| } // bridgeContext != null |
| else { |
| event = EntryEventImpl.create(r, getOperation(), getKey(), null, /* newValue */ |
| getCallbackArg(), this.useOriginRemote, eventSender, true/* generateCallbacks */, |
| false/* initializeId */); |
| } |
| |
| event.setCausedByMessage(this); |
| |
| if (this.versionTag != null) { |
| this.versionTag.replaceNullIDs(getSender()); |
| event.setVersionTag(this.versionTag); |
| } |
| // for cq processing and client notification by BS. |
| if (this.hasOldValue) { |
| if (this.oldValueIsSerialized) { |
| event.setSerializedOldValue(getOldValueBytes()); |
| } else { |
| event.setOldValue(getOldValueBytes()); |
| } |
| } |
| |
| Assert.assertTrue(eventId != null); |
| event.setEventId(eventId); |
| |
| event.setPossibleDuplicate(this.possibleDuplicate); |
| |
| try { |
| r.getDataView().destroyOnRemote(event, true, this.expectedOldValue); |
| sendReply(dm, event.getVersionTag()); |
| } catch (CacheWriterException cwe) { |
| sendReply(getSender(), this.processorId, dm, new ReplyException(cwe), r, startTime); |
| return false; |
| } catch (EntryNotFoundException eee) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("operateOnRegion caught EntryNotFoundException", eee); |
| } |
| ReplyMessage.send(getSender(), getProcessorId(), new ReplyException(eee), |
| getReplySender(dm), r.isInternalRegion()); |
| } catch (DataLocationException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("operateOnRegion caught DataLocationException"); |
| } |
| ReplyMessage.send(getSender(), getProcessorId(), new ReplyException(e), getReplySender(dm), |
| r.isInternalRegion()); |
| } |
| return false; |
| } finally { |
| if (event != null) { |
| event.release(); |
| } |
| } |
| } |
| |
| @Override |
| public int getDSFID() { |
| return R_DESTROY_MESSAGE; |
| } |
| |
| private void sendReply(DistributionManager dm, VersionTag<?> versionTag) { |
| DestroyReplyMessage.send(this.getSender(), getReplySender(dm), this.processorId, versionTag); |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| setKey(DataSerializer.readObject(in)); |
| this.cbArg = DataSerializer.readObject(in); |
| this.op = Operation.fromOrdinal(in.readByte()); |
| if ((flags & HAS_BRIDGE_CONTEXT) != 0) { |
| this.bridgeContext = DataSerializer.readObject(in); |
| } |
| if ((flags & HAS_ORIGINAL_SENDER) != 0) { |
| this.originalSender = DataSerializer.readObject(in); |
| } |
| this.eventId = DataSerializer.readObject(in); |
| |
| // for old values for CQs |
| if (this.hasOldValue) { |
| // out.writeBoolean(this.hasOldValue); |
| // below boolean is not strictly required, but this is for compatibility |
| in.readByte(); |
| setOldValBytes(DataSerializer.readByteArray(in)); |
| } |
| this.expectedOldValue = DataSerializer.readObject(in); |
| // to prevent bug 51024 always call readObject for versionTag |
| // since toData always calls writeObject for versionTag. |
| this.versionTag = DataSerializer.readObject(in); |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| super.toData(out, context); |
| DataSerializer.writeObject(getKey(), out); |
| DataSerializer.writeObject(this.cbArg, out); |
| out.writeByte(this.op.ordinal); |
| if (this.bridgeContext != null) { |
| DataSerializer.writeObject(this.bridgeContext, out); |
| } |
| if (this.originalSender != null) { |
| DataSerializer.writeObject(this.originalSender, out); |
| } |
| DataSerializer.writeObject(this.eventId, out); |
| |
| // this will be on wire for cqs old value generations. |
| if (this.hasOldValue) { |
| out.writeByte(this.oldValueIsSerialized ? 1 : 0); |
| byte policy = DistributedCacheOperation.valueIsToDeserializationPolicy(oldValueIsSerialized); |
| DistributedCacheOperation.writeValue(policy, getOldValObj(), getOldValueBytes(), out); |
| } |
| DataSerializer.writeObject(this.expectedOldValue, out); |
| DataSerializer.writeObject(this.versionTag, out); |
| } |
| |
| @Override |
| protected void setFlags(short flags, DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.setFlags(flags, in, context); |
| this.hasOldValue = (flags & HAS_OLD_VALUE) != 0; |
| this.useOriginRemote = (flags & USE_ORIGIN_REMOTE) != 0; |
| this.possibleDuplicate = (flags & POS_DUP) != 0; |
| } |
| |
| @Override |
| protected short computeCompressedShort() { |
| short s = super.computeCompressedShort(); |
| // this will be on wire for cqs old value generations. |
| if (this.hasOldValue) |
| s |= HAS_OLD_VALUE; |
| if (this.useOriginRemote) |
| s |= USE_ORIGIN_REMOTE; |
| if (this.possibleDuplicate) |
| s |= POS_DUP; |
| if (this.bridgeContext != null) |
| s |= HAS_BRIDGE_CONTEXT; |
| if (this.originalSender != null) |
| s |= HAS_ORIGINAL_SENDER; |
| if (this.versionTag != null) |
| s |= HAS_VERSION_TAG; |
| return s; |
| } |
| |
| @Override |
| public EventID getEventID() { |
| return this.eventId; |
| } |
| |
| /** |
| * Assists the toString method in reporting the contents of this message |
| * |
| */ |
| @Override |
| protected void appendFields(StringBuffer buff) { |
| super.appendFields(buff); |
| buff.append("; key=").append(getKey()); |
| if (originalSender != null) { |
| buff.append("; originalSender=").append(originalSender); |
| } |
| if (bridgeContext != null) { |
| buff.append("; bridgeContext=").append(bridgeContext); |
| } |
| if (eventId != null) { |
| buff.append("; eventId=").append(eventId); |
| } |
| buff.append("; hasOldValue= ").append(this.hasOldValue); |
| } |
| |
| protected Object getKey() { |
| return this.key; |
| } |
| |
| private void setKey(Object key) { |
| this.key = key; |
| } |
| |
| public Operation getOperation() { |
| return this.op; |
| } |
| |
| protected Object getCallbackArg() { |
| return this.cbArg; |
| } |
| |
| @Override |
| public boolean prefersOldSerialized() { |
| return true; |
| } |
| |
| @Override |
| public boolean isUnretainedOldReferenceOk() { |
| return true; |
| } |
| |
| @Override |
| public boolean isCachedDeserializableValueOk() { |
| return false; |
| } |
| |
| private void setOldValueIsSerialized(boolean isSerialized) { |
| if (isSerialized) { |
| // Defer serialization until toData is called. |
| this.oldValueIsSerialized = true; // VALUE_IS_SERIALIZED_OBJECT; |
| } else { |
| this.oldValueIsSerialized = false; // VALUE_IS_BYTES; |
| } |
| } |
| |
| @Override |
| public void importOldObject(@Unretained(ENTRY_EVENT_OLD_VALUE) Object ov, boolean isSerialized) { |
| setOldValueIsSerialized(isSerialized); |
| // Defer serialization until toData is called. |
| setOldValObj(ov); |
| } |
| |
| @Override |
| public void importOldBytes(byte[] ov, boolean isSerialized) { |
| setOldValueIsSerialized(isSerialized); |
| setOldValBytes(ov); |
| } |
| |
| public static class DestroyReplyMessage extends ReplyMessage { |
| |
| private static final byte HAS_VERSION = 0x01; |
| private static final byte PERSISTENT = 0x02; |
| |
| private VersionTag<?> versionTag; |
| |
| /** DSFIDFactory constructor */ |
| public DestroyReplyMessage() {} |
| |
| static void send(InternalDistributedMember recipient, ReplySender dm, int procId, |
| VersionTag<?> versionTag) { |
| Assert.assertTrue(recipient != null, "DestroyReplyMessage NULL recipient"); |
| DestroyReplyMessage m = new DestroyReplyMessage(recipient, procId, versionTag); |
| dm.putOutgoing(m); |
| } |
| |
| DestroyReplyMessage(InternalDistributedMember recipient, int procId, VersionTag<?> versionTag) { |
| this.setProcessorId(procId); |
| this.setRecipient(recipient); |
| this.versionTag = versionTag; |
| } |
| |
| @Override |
| public boolean getInlineProcess() { |
| return true; |
| } |
| |
| @Override |
| public int getDSFID() { |
| return R_DESTROY_REPLY_MESSAGE; |
| } |
| |
| @Override |
| public void process(final DistributionManager dm, final ReplyProcessor21 rp) { |
| final long startTime = getTimestamp(); |
| if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) { |
| logger.trace(LogMarker.DM_VERBOSE, |
| "DestroyReplyMessage process invoking reply processor with processorId:{}", |
| this.processorId); |
| } |
| if (rp == null) { |
| if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) { |
| logger.trace(LogMarker.DM_VERBOSE, "DestroyReplyMessage processor not found"); |
| } |
| return; |
| } |
| if (this.versionTag != null) { |
| this.versionTag.replaceNullIDs(getSender()); |
| } |
| if (rp instanceof RemoteDestroyReplyProcessor) { |
| RemoteDestroyReplyProcessor processor = (RemoteDestroyReplyProcessor) rp; |
| processor.setResponse(this.versionTag); |
| } |
| rp.process(this); |
| |
| if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) { |
| logger.trace(LogMarker.DM_VERBOSE, "{} processed {}", rp, this); |
| } |
| dm.getStats().incReplyMessageTime(NanoTimer.getTime() - startTime); |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| super.toData(out, context); |
| byte b = 0; |
| if (this.versionTag != null) { |
| b |= HAS_VERSION; |
| } |
| if (this.versionTag instanceof DiskVersionTag) { |
| b |= PERSISTENT; |
| } |
| out.writeByte(b); |
| if (this.versionTag != null) { |
| InternalDataSerializer.invokeToData(this.versionTag, out); |
| } |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| byte b = in.readByte(); |
| boolean hasTag = (b & HAS_VERSION) != 0; |
| boolean persistentTag = (b & PERSISTENT) != 0; |
| if (hasTag) { |
| this.versionTag = VersionTag.create(persistentTag, in); |
| } |
| } |
| |
| @Override |
| public String toString() { |
| StringBuilder sb = super.getStringBuilder(); |
| sb.append(getShortClassName()); |
| sb.append(" processorId="); |
| sb.append(this.processorId); |
| if (this.versionTag != null) { |
| sb.append(" version=").append(this.versionTag); |
| } |
| sb.append(" from "); |
| sb.append(this.getSender()); |
| ReplyException ex = getException(); |
| if (ex != null) { |
| sb.append(" with exception "); |
| sb.append(ex); |
| } |
| return sb.toString(); |
| } |
| |
| } |
| static class RemoteDestroyReplyProcessor extends RemoteOperationResponse { |
| VersionTag<?> versionTag; |
| |
| RemoteDestroyReplyProcessor(InternalDistributedSystem ds, DistributedMember recipient, |
| Object key) { |
| super(ds, (InternalDistributedMember) recipient, false); |
| } |
| |
| void setResponse(VersionTag<?> versionTag) { |
| this.versionTag = versionTag; |
| } |
| |
| VersionTag<?> getVersionTag() { |
| return this.versionTag; |
| } |
| } |
| } |
