@Override
    protected void process(final ClusterDistributionManager dm) {
      Throwable thr = null;
      ReplyException rex = null;
      Object nextObject = null;
      Object failedObject = null;
      int socketBufferSize = dm.getSystem().getConfig().getSocketBufferSize();
      int chunkSize = socketBufferSize - MSG_OVERHEAD;
      HeapDataOutputStream outStream =
          new HeapDataOutputStream(chunkSize, Versioning
              .getKnownVersionOrDefault(getSender().getVersion(),
                  KnownVersion.CURRENT));
      boolean sentFinalMessage = false;
      boolean receiverCacheClosed = false;
      int msgNum = 0;

      try {
        do {
          int numObjectsInChunk = 0;
          // boolean firstObject = true;

          // always write at least one object, allowing expansion
          // if we have an object already that didn't get added, then use
          // that object instead of getting another one
          if (failedObject == null) {
            nextObject = getNextReplyObject();
          } else {
            nextObject = failedObject;
            failedObject = null;
          }

          if (nextObject != Token.END_OF_STREAM) {
            numObjectsInChunk = 1;
            BlobHelper.serializeTo(nextObject, outStream);

            // for the next objects, disallow stream from allocating more storage
            do {
              outStream.disallowExpansion(CHUNK_FULL); // sets the mark where rollback occurs on
                                                       // CHUNK_FULL

              nextObject = getNextReplyObject();

              if (nextObject != Token.END_OF_STREAM) {
                try {
                  BlobHelper.serializeTo(nextObject, outStream);
                  numObjectsInChunk++;
                } catch (GemFireRethrowable e) {
                  // can only be thrown when expansion is disallowed
                  // and buffer is automatically reset to point where it was disallowed
                  failedObject = nextObject;
                  break;
                }
              }
            } while (nextObject != Token.END_OF_STREAM);
          }

          try {
            replyWithData(dm, outStream, numObjectsInChunk, msgNum++,
                nextObject == Token.END_OF_STREAM);
            if (nextObject == Token.END_OF_STREAM) {
              sentFinalMessage = true;
            }
          } catch (CancelException e) {
            receiverCacheClosed = true;
            break; // receiver no longer cares
          }
          outStream.reset(); // ready for reuse, assumes replyWithData
                             // does not queue the message but outStream has
                             // already been used
        } while (nextObject != Token.END_OF_STREAM);
        // } catch (CancelException e) {
        // // if cache is closed, we cannot send a reply (correct?)
      } catch (VirtualMachineError err) {
        SystemFailure.initiateFailure(err);
        // If this ever returns, rethrow the error. We're poisoned
        // now, so don't let this thread continue.
        throw err;
      } catch (Throwable t) {
        // Whenever you catch Error or Throwable, you must also
        // catch VirtualMachineError (see above). However, there is
        // _still_ a possibility that you are dealing with a cascading
        // error condition, so you also need to check to see if the JVM
        // is still usable:
        SystemFailure.checkFailure();
        thr = t;
      }

      if (thr != null) {
        rex = new ReplyException(thr);
        replyWithException(dm, rex);
      } else if (!sentFinalMessage && !receiverCacheClosed) {
        throw new InternalGemFireError(
            "this should not happen");
        // replyNoData(dm);
      }
    }
Overridden to wait for messages being currently processed: This situation can come about if a
member departs while we are still processing data from that member
geode-core/src/main/java/org/apache/geode/distributed/internal/streaming/StreamingOperation.java