salesforce-lib/src/main/java/com/streamsets/pipeline/stage/origin/salesforce/ForceSource.java
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    String nextSourceOffset = null;

    LOG.debug("lastSourceOffset: {}", lastSourceOffset);

    // send event only once for each time we run out of data.
    if(shouldSendNoMoreDataEvent) {
      CommonEvents.NO_MORE_DATA.create(getContext()).createAndSend();
      shouldSendNoMoreDataEvent = false;
      return lastSourceOffset;
    }

    int batchSize = Math.min(conf.basicConfig.maxBatchSize, maxBatchSize);

    if (!conf.queryExistingData ||
        (null != lastSourceOffset && lastSourceOffset.startsWith(EVENT_ID_OFFSET_PREFIX))) {
      if (conf.subscribeToStreaming) {
        nextSourceOffset = streamingProduce(lastSourceOffset, batchSize, batchMaker);
      } else {
        // We're done reading existing data, but we don't want to subscribe to Streaming API
        return null;
      }
    } else if (conf.queryExistingData) {
      if (!queryInProgress()) {
        long now = System.currentTimeMillis();
        long delay = Math.max(0, (lastQueryCompletedTime + (1000 * conf.queryInterval)) - now);

        if (delay > 0) {
          // Sleep in one second increments so we don't tie up the app.
          LOG.info("{}ms remaining until next fetch.", delay);
          ThreadUtil.sleep(Math.min(delay, 1000));

          return lastSourceOffset;
        }
      }

      if (conf.useBulkAPI) {
        nextSourceOffset = bulkProduce(lastSourceOffset, batchSize, batchMaker);
      } else {
        nextSourceOffset = soapProduce(lastSourceOffset, batchSize, batchMaker);
      }
    } else if (conf.subscribeToStreaming) {
      // No offset, and we're not querying existing data, so switch to streaming
      nextSourceOffset = READ_EVENTS_FROM_NOW;
    }

    LOG.debug("nextSourceOffset: {}", nextSourceOffset);

    return nextSourceOffset;
  }
@Override
  public String produce(String lastSourceOffset, int maxBatchSize, BatchMaker batchMaker) throws StageException {
    String nextSourceOffset = null;

    LOG.debug("lastSourceOffset: {}", lastSourceOffset);

    // send event only once for each time we run out of data.
    if(shouldSendNoMoreDataEvent) {
      CommonEvents.NO_MORE_DATA.create(getContext()).createAndSend();
      shouldSendNoMoreDataEvent = false;
      return lastSourceOffset;
    }

    int batchSize = Math.min(conf.basicConfig.maxBatchSize, maxBatchSize);

    if (!conf.queryExistingData ||
        (null != lastSourceOffset && lastSourceOffset.startsWith(EVENT_ID_OFFSET_PREFIX))) {
      if (conf.subscribeToStreaming) {
        nextSourceOffset = streamingProduce(lastSourceOffset, batchSize, batchMaker);
      } else {
        // We're done reading existing data, but we don't want to subscribe to Streaming API
        return null;
      }
    } else if (conf.queryExistingData) {
      if (!queryInProgress()) {
        long now = System.currentTimeMillis();
        long delay = Math.max(0, (lastQueryCompletedTime + (1000 * conf.queryInterval)) - now);

        if (delay > 0) {
          // Sleep in one second increments so we don't tie up the app.
          LOG.info("{}ms remaining until next fetch.", delay);
          ThreadUtil.sleep(Math.min(delay, 1000));

          return lastSourceOffset;
        }
      }

      if (conf.useBulkAPI) {
        nextSourceOffset = bulkProduce(lastSourceOffset, batchSize, batchMaker);
      } else {
        nextSourceOffset = soapProduce(lastSourceOffset, batchSize, batchMaker);
      }
    } else if (conf.subscribeToStreaming) {
      // No offset, and we're not querying existing data, so switch to streaming
      nextSourceOffset = READ_EVENTS_FROM_NOW;
    }

    LOG.debug("nextSourceOffset: {}", nextSourceOffset);

    return nextSourceOffset;
  }
@ Override public String produce ( String lastSourceOffset , int maxBatchSize , BatchMaker batchMaker ) throws StageException { String nextSourceOffset = null ; LOG . debug ( "lastSourceOffset: {}" , lastSourceOffset ) ; // send event only once for each time we run out of data. if ( shouldSendNoMoreDataEvent ) { CommonEvents . NO_MORE_DATA . create ( getContext ( ) ) . createAndSend ( ) ; shouldSendNoMoreDataEvent = false ; return lastSourceOffset ; } int batchSize = Math . min ( conf . basicConfig . maxBatchSize , maxBatchSize ) ; if ( ! conf . queryExistingData || ( null != lastSourceOffset && lastSourceOffset . startsWith ( EVENT_ID_OFFSET_PREFIX ) ) ) { if ( conf . subscribeToStreaming ) { nextSourceOffset = streamingProduce ( lastSourceOffset , batchSize , batchMaker ) ; } else { // We're done reading existing data, but we don't want to subscribe to Streaming API return null ; } } else if ( conf . queryExistingData ) { if ( ! queryInProgress ( ) ) { long now = System . currentTimeMillis ( ) ; long delay = Math . max ( 0 , ( lastQueryCompletedTime + ( 1000 * conf . queryInterval ) ) - now ) ; if ( delay > 0 ) { // Sleep in one second increments so we don't tie up the app. LOG . info ( "{}ms remaining until next fetch." , delay ) ; ThreadUtil . sleep ( Math . min ( delay , 1000 ) ) ; return lastSourceOffset ; } } if ( conf . useBulkAPI ) { nextSourceOffset = bulkProduce ( lastSourceOffset , batchSize , batchMaker ) ; } else { nextSourceOffset = soapProduce ( lastSourceOffset , batchSize , batchMaker ) ; } } else if ( conf . subscribeToStreaming ) { // No offset, and we're not querying existing data, so switch to streaming nextSourceOffset = READ_EVENTS_FROM_NOW ; } LOG . debug ( "nextSourceOffset: {}" , nextSourceOffset ) ; return nextSourceOffset ; }
{@inheritDoc}
{