public void run() {
      try {
         // TODO: need to change to SDC.forEachSegment
         // send data container entries
         for (InternalCacheEntry ice : dataContainer) {
            Object key = ice.getKey();  //todo [anistor] should we check for expired entries?
            int segmentId = keyPartitioner.getSegment(key);
            if (segments.contains(segmentId) && !ice.isL1Entry()) {
               InternalCacheEntry entry = mapEntryFromDataContainer.apply(ice, entryFactory);
               if (entry != null) {
                  sendEntry(entry, segmentId);
               }
            }
         }

         AdvancedCacheLoader<Object, Object> stProvider = persistenceManager.getStateTransferProvider();
         if (stProvider != null) {
            try {
               Flowable.fromPublisher(stProvider.entryPublisher(k -> !dataContainer.containsKey(k), true, true))
                     .blockingForEach(me -> {
                        int segmentId = keyPartitioner.getSegment(me.getKey());
                        if (segments.contains(segmentId)) {
                           try {
                              InternalCacheEntry entry = mapEntryFromStore.apply(me, entryFactory);
                              if (entry != null) {
                                 sendEntry(entry, segmentId);
                              }
                           } catch (CacheException e) {
                              log.failedLoadingValueFromCacheStore(me.getKey(), e);
                           }
                        }
                     });
            } catch (CacheException e) {
               log.failedLoadingKeysFromCacheStore(e);
            }
         }

         // send the last chunk of all segments
         sendEntries(true);
      } catch (Throwable t) {
         // ignore eventual exceptions caused by cancellation (have InterruptedException as the root cause)
         if (isCancelled()) {
            if (trace) {
               log.tracef("Ignoring error in already cancelled transfer to node %s, segments %s", destination,
                          segments);
            }
         } else {
            log.failedOutBoundTransferExecution(t);
         }
      }
      if (trace) {
         log.tracef("Completed outbound transfer to node %s, segments %s", destination, segments);
      }
   }
public void run ( ) { try { for ( InternalCacheEntry ice : dataContainer ) { Object key = ice . getKey ( ) ; int segmentId = keyPartitioner . getSegment ( key ) ; if ( segments . contains ( segmentId ) && ! ice . isL1Entry ( ) ) { InternalCacheEntry entry = mapEntryFromDataContainer . apply ( ice , entryFactory ) ; if ( entry != null ) { sendEntry ( entry , segmentId ) ; } } } AdvancedCacheLoader < Object , Object > stProvider = persistenceManager . getStateTransferProvider ( ) ; if ( stProvider != null ) { try { Flowable . fromPublisher ( stProvider . entryPublisher ( k -> ! dataContainer . containsKey ( k ) , true , true ) ) . blockingForEach ( me -> { int segmentId = keyPartitioner . getSegment ( me . getKey ( ) ) ; if ( segments . contains ( segmentId ) ) { try { InternalCacheEntry entry = mapEntryFromStore . apply ( me , entryFactory ) ; if ( entry != null ) { sendEntry ( entry , segmentId ) ; } } catch ( CacheException e ) { log . failedLoadingValueFromCacheStore ( me . getKey ( ) , e ) ; } } } ) ; } catch ( CacheException e ) { log . failedLoadingKeysFromCacheStore ( e ) ; } } sendEntries ( true ) ; } catch ( Throwable t ) { if ( isCancelled ( ) ) { if ( trace ) { log . tracef ( "Ignoring error in already cancelled transfer to node %s, segments %s" , destination , segments ) ; } } else { log . failedOutBoundTransferExecution ( t ) ; } } if ( trace ) { log . tracef ( "Completed outbound transfer to node %s, segments %s" , destination , segments ) ; } }
todo [anistor] check thread interrupt status in loops to implement faster cancellation