DistributedRegionFunctionStreamingMessage extends DistributionMessage implements TransactionMessage { protected boolean operateOnDistributedRegion(final ClusterDistributionManager dm, DistributedRegion r) throws ForceReattemptException { if (this.functionObject == null) { ReplyMessage.send(getSender(), this.processorId, new ReplyException(new FunctionException( String.format("Function named %s is not registered to FunctionService", this.functionName))), dm, r.isInternalRegion()); return false; } if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) { logger.trace(LogMarker.DM_VERBOSE, "FunctionMessage operateOnRegion: {}", r.getFullPath()); } try { r.executeOnRegion(this, this.functionObject, this.args, this.processorId, this.filter, this.isReExecute); if (!this.replyLastMsg && this.functionObject.hasResult()) { ReplyMessage.send(getSender(), this.processorId, new ReplyException(new FunctionException( String.format("The function, %s, did not send last result", functionObject.getId()))), dm, r.isInternalRegion()); return false; } } catch (IOException e) { ReplyMessage .send(getSender(), this.processorId, new ReplyException( "Operation got interrupted due to shutdown in progress on remote VM", e), dm, r.isInternalRegion()); return false; } catch (CancelException sde) { ReplyMessage.send(getSender(), this.processorId, new ReplyException(new ForceReattemptException( "Operation got interrupted due to shutdown in progress on remote VM", sde)), dm, r.isInternalRegion()); return false; } return false; } DistributedRegionFunctionStreamingMessage();  DistributedRegionFunctionStreamingMessage(final String regionPath, Function function,
      int procId, final Set filter, Object args, boolean isReExecute,
      boolean isFnSerializationReqd); @Override int getProcessorId(); @Override int getDSFID(); @Override void fromData(DataInput in,
      DeserializationContext context); @Override void toData(DataOutput out,
      SerializationContext context); synchronized boolean sendReplyForOneResult(DistributionManager dm, Object oneResult,
      boolean lastResult, boolean sendResultsInOrder); @Override int getProcessorType(); @Override boolean canStartRemoteTransaction(); @Override int getTXUniqId(); @Override InternalDistributedMember getMemberToMasqueradeAs(); @Override InternalDistributedMember getTXOriginatorClient(); @Override boolean canParticipateInTransaction(); @Override boolean isTransactionDistributed();  }
@Test public void shouldBeMockable() throws Exception { DistributedRegionFunctionStreamingMessage mockDistributedRegionFunctionStreamingMessage = mock(DistributedRegionFunctionStreamingMessage.class); ClusterDistributionManager mockDistributionManager = mock(ClusterDistributionManager.class); DistributedRegion mockDistributedRegion = mock(DistributedRegion.class); when(mockDistributedRegionFunctionStreamingMessage .operateOnDistributedRegion(eq(mockDistributionManager), eq(mockDistributedRegion))) .thenReturn(true); assertThat(mockDistributedRegionFunctionStreamingMessage .operateOnDistributedRegion(mockDistributionManager, mockDistributedRegion)).isTrue(); }