/** 
     * Send an ack 
     * @throws ForceReattemptException if the peer is no longer available
     */
public static void send(final InternalDistributedMember recipient, final int processorId,
        final DM dm, Set keys) throws ForceReattemptException {
      Assert.assertTrue(recipient != null, "FetchKeysReplyMessage NULL reply message");
      final int numSeries = 1;
      final int seriesNum = 0;
      // chunkEntries returns false if didn't finish
      if (logger.isDebugEnabled()) {
        logger.debug("Starting pr keys chunking for {} kets to member {}", keys.size(), recipient);
      }
      try {
        boolean finished = chunkSet(recipient, keys, InitialImageOperation.CHUNK_SIZE_IN_BYTES, false,
          new ObjectIntProcedure() {
            int msgNum = 0;
            boolean last = false;
            /**
              * @param a byte[] chunk
              * @param b positive if last chunk
              * @return true to continue to next chunk
              */
            public boolean executeWith(Object a, int b) {
//              if (this.last)
//                throw new InternalGemFireError(LocalizedStrings.FetchKeysMessage_ALREADY_PROCESSED_LAST_CHUNK.toLocalizedString());
              HeapDataOutputStream chunk = (HeapDataOutputStream)a;
              this.last = b > 0;
              try {
                boolean okay = sendChunk(recipient, processorId, dm, chunk, seriesNum, msgNum++, numSeries, this.last);
                return okay;
              }
              catch (CancelException e) {
                return false;
              }
            }
          });
        if (logger.isDebugEnabled()) {
          logger.debug("{} pr keys chunking", (finished?"Finished" : "DID NOT complete"));
        }
      }
      catch (IOException io) {
        throw new ForceReattemptException(LocalizedStrings.FetchKeysMessage_UNABLE_TO_SEND_RESPONSE_TO_FETCH_KEYS_REQUEST.toLocalizedString(), io);
      }
      // TODO [bruce] pass a reference to the cache or region down here so we can do this test
      //Assert.assertTrue(!cache is closed, "chunking interrupted but cache is still open");
    }
public static void send(final InternalDistributedMember recipient, final int processorId,
final DM dm, Set keys) throws ForceReattemptException {
Assert.assertTrue(recipient != null, "FetchKeysReplyMessage NULL reply message");
final int numSeries = 1;
final int seriesNum = 0;
// chunkEntries returns false if didn't finish
if (logger.isDebugEnabled()) {
logger.debug("Starting pr keys chunking for {} kets to member {}", keys.size(), recipient);
}
try {
boolean finished = chunkSet(recipient, keys, InitialImageOperation.CHUNK_SIZE_IN_BYTES, false,
new ObjectIntProcedure() {
int msgNum = 0;
boolean last = false;
/**
* @param a byte[] chunk
* @param b positive if last chunk
* @return true to continue to next chunk
*/
public boolean executeWith(Object a, int b) {
// if (this.last)
// throw new InternalGemFireError(LocalizedStrings.FetchKeysMessage_ALREADY_PROCESSED_LAST_CHUNK.toLocalizedString());
HeapDataOutputStream chunk = (HeapDataOutputStream)a;
this.last = b > 0;
try {
boolean okay = sendChunk(recipient, processorId, dm, chunk, seriesNum, msgNum++, numSeries, this.last);
return okay;
}
catch (CancelException e) {
return false;
}
}
});
if (logger.isDebugEnabled()) {
logger.debug("{} pr keys chunking", (finished?"Finished" : "DID NOT complete"));
}
}
catch (IOException io) {
throw new ForceReattemptException(LocalizedStrings.FetchKeysMessage_UNABLE_TO_SEND_RESPONSE_TO_FETCH_KEYS_REQUEST.toLocalizedString(), io);
}
// TODO [bruce] pass a reference to the cache or region down here so we can do this test
//Assert.assertTrue(!cache is closed, "chunking interrupted but cache is still open");
}
public static void send(final InternalDistributedMember recipient, final int processorId,
final DM dm, Set keys) throws ForceReattemptException {
Assert.assertTrue(recipient != null, "FetchKeysReplyMessage NULL reply message");
final int numSeries = 1;
final int seriesNum = 0;
// chunkEntries returns false if didn't finish
if (logger.isDebugEnabled()) {
logger.debug("Starting pr keys chunking for {} kets to member {}", keys.size(), recipient);
}
try {
boolean finished = chunkSet(recipient, keys, InitialImageOperation.CHUNK_SIZE_IN_BYTES, false,
new ObjectIntProcedure() {
int msgNum = 0;
boolean last = false;
/**
* @param a byte[] chunk
* @param b positive if last chunk
* @return true to continue to next chunk
*/
public boolean executeWith(Object a, int b) {
// if (this.last)
// throw new InternalGemFireError(LocalizedStrings.FetchKeysMessage_ALREADY_PROCESSED_LAST_CHUNK.toLocalizedString());
HeapDataOutputStream chunk = (HeapDataOutputStream)a;
this.last = b > 0;
try {
boolean okay = sendChunk(recipient, processorId, dm, chunk, seriesNum, msgNum++, numSeries, this.last);
return okay;
}
catch (CancelException e) {
return false;
}
}
});
if (logger.isDebugEnabled()) {
logger.debug("{} pr keys chunking", (finished?"Finished" : "DID NOT complete"));
}
}
catch (IOException io) {
throw new ForceReattemptException(LocalizedStrings.FetchKeysMessage_UNABLE_TO_SEND_RESPONSE_TO_FETCH_KEYS_REQUEST.toLocalizedString(), io);
}
// TODO [bruce] pass a reference to the cache or region down here so we can do this test
//Assert.assertTrue(!cache is closed, "chunking interrupted but cache is still open");
}
public static void send(final InternalDistributedMember recipient, final int processorId,
final DM dm, Set keys) throws ForceReattemptException {
Assert.assertTrue(recipient != null, "FetchKeysReplyMessage NULL reply message");
final int numSeries = 1;
final int seriesNum = 0;
// chunkEntries returns false if didn't finish
if (logger.isDebugEnabled()) {
logger.debug("Starting pr keys chunking for {} kets to member {}", keys.size(), recipient);
}
try {
boolean finished = chunkSet(recipient, keys, InitialImageOperation.CHUNK_SIZE_IN_BYTES, false,
new ObjectIntProcedure() {
int msgNum = 0;
boolean last = false;
/**
* @param a byte[] chunk
* @param b positive if last chunk
* @return true to continue to next chunk
*/
public boolean executeWith(Object a, int b) {
// if (this.last)
// throw new InternalGemFireError(LocalizedStrings.FetchKeysMessage_ALREADY_PROCESSED_LAST_CHUNK.toLocalizedString());
HeapDataOutputStream chunk = (HeapDataOutputStream)a;
this.last = b > 0;
try {
boolean okay = sendChunk(recipient, processorId, dm, chunk, seriesNum, msgNum++, numSeries, this.last);
return okay;
}
catch (CancelException e) {
return false;
}
}
});
if (logger.isDebugEnabled()) {
logger.debug("{} pr keys chunking", (finished?"Finished" : "DID NOT complete"));
}
}
catch (IOException io) {
throw new ForceReattemptException(LocalizedStrings.FetchKeysMessage_UNABLE_TO_SEND_RESPONSE_TO_FETCH_KEYS_REQUEST.toLocalizedString(), io);
}
// TODO [bruce] pass a reference to the cache or region down here so we can do this test
//Assert.assertTrue(!cache is closed, "chunking interrupted but cache is still open");
}