content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements.  See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License.  You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.camel.component.mybatis;
  
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Queue;
  
  import org.apache.camel.Exchange;
  import org.apache.camel.ExchangePattern;
  import org.apache.camel.ExchangePropertyKey;
  import org.apache.camel.Message;
  import org.apache.camel.Processor;
  import org.apache.camel.RollbackExchangeException;
  import org.apache.camel.ShutdownRunningTask;
  import org.apache.camel.support.ScheduledBatchPollingConsumer;
  import org.apache.camel.util.CastUtils;
  import org.apache.camel.util.ObjectHelper;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Consumer to read data from a database.
   */
  public class MyBatisConsumer extends ScheduledBatchPollingConsumer {
  
      static final class DataHolder {
          Exchange exchange;
          Object data;
  
          DataHolder() {
          }
      }
  
      private static final Logger LOG = LoggerFactory.getLogger(MyBatisConsumer.class);
  
      protected volatile ShutdownRunningTask shutdownRunningTask;
      protected volatile int pendingExchanges;
  
      private String onConsume;
      private boolean useIterator = true;
      private boolean routeEmptyResultSet;
  
      public MyBatisConsumer(MyBatisEndpoint endpoint, Processor processor) {
          super(endpoint, processor);
      }
  
      @Override
      public MyBatisEndpoint getEndpoint() {
          return (MyBatisEndpoint) super.getEndpoint();
      }
  
      /**
       * Polls the database
       */
      @Override
      protected int poll() throws Exception {
          // must reset for each poll
          shutdownRunningTask = null;
          pendingExchanges = 0;
  
          // poll data from the database
          MyBatisEndpoint endpoint = getEndpoint();
          LOG.trace("Polling: {}", endpoint);
          List<?> data = endpoint.getProcessingStrategy().poll(this, getEndpoint());
  
          // okay we have some response from MyBatis so lets mark the consumer as ready
          forceConsumerAsReady();
  
          // create a list of exchange objects with the data
          Queue<DataHolder> answer = new LinkedList<>();
          if (useIterator) {
              for (Object item : data) {
                  Exchange exchange = createExchange(item);
                  DataHolder holder = new DataHolder();
                  holder.exchange = exchange;
                  holder.data = item;
                  answer.add(holder);
              }
          } else {
              if (!data.isEmpty() || routeEmptyResultSet) {
                  Exchange exchange = createExchange(data);
                  DataHolder holder = new DataHolder();
                  holder.exchange = exchange;
                  holder.data = data;
                  answer.add(holder);
              }
          }
  
          // process all the exchanges in this batch
          return processBatch(CastUtils.cast(answer));
      }
  
      @Override
      public int processBatch(Queue<Object> exchanges) throws Exception {
          final MyBatisEndpoint endpoint = getEndpoint();
  
          int total = exchanges.size();
  
          // limit if needed
          if (maxMessagesPerPoll > 0 && total > maxMessagesPerPoll) {
              LOG.debug("Limiting to maximum messages to poll {} as there were {} messages in this poll.",
                      maxMessagesPerPoll, total);
              total = maxMessagesPerPoll;
          }
  
          for (int index = 0; index < total && isBatchAllowed(); index++) {
              // only loop if we are started (allowed to run)
              DataHolder holder = ObjectHelper.cast(DataHolder.class, exchanges.poll());
              Exchange exchange = holder.exchange;
              Object data = holder.data;
  
              // add current index and total as properties
              exchange.setProperty(ExchangePropertyKey.BATCH_INDEX, index);
              exchange.setProperty(ExchangePropertyKey.BATCH_SIZE, total);
              exchange.setProperty(ExchangePropertyKey.BATCH_COMPLETE, index == total - 1);
  
              // update pending number of exchanges
              pendingExchanges = total - index - 1;
  
              Exception cause = null;
              try {
                  getProcessor().process(exchange);
  
                  if (onConsume != null) {
                      endpoint.getProcessingStrategy().commit(endpoint, exchange, data, onConsume);
                  }
              } catch (Exception e) {
                  handleException(e);
              }
  
              if (getEndpoint().isTransacted() && exchange.isFailed()) {
                  // break out as we are transacted and should rollback
                  cause = exchange.getException();
                  if (cause == null) {
                      cause = new RollbackExchangeException("Rollback transaction due error processing exchange", null);
                  }
              }
              releaseExchange(exchange, false);
  
              if (cause != null) {
                  throw cause;
              }
          }
  
          return total;
      }
  
      private Exchange createExchange(Object data) {
          final MyBatisEndpoint endpoint = getEndpoint();
          final Exchange exchange = createExchange(false);
          exchange.setPattern(ExchangePattern.InOnly);
          final String outputHeader = getEndpoint().getOutputHeader();
  
          Message msg = exchange.getIn();
          if (outputHeader != null) {
              msg.setHeader(outputHeader, data);
          } else {
              msg.setBody(data);
          }
          msg.setHeader(MyBatisConstants.MYBATIS_STATEMENT_NAME, endpoint.getStatement());
  
          return exchange;
      }
  
      /**
       * Gets the statement(s) to run after successful processing. Use comma to separate multiple statements.
       */
      public String getOnConsume() {
          return onConsume;
      }
  
      /**
       * Sets the statement to run after successful processing. Use comma to separate multiple statements.
       */
      public void setOnConsume(String onConsume) {
          this.onConsume = onConsume;
      }
  
      /**
       * Indicates how resultset should be delivered to the route
       */
      public boolean isUseIterator() {
          return useIterator;
      }
  
      /**
       * Sets how resultset should be delivered to route. Indicates delivery as either a list or individual object.
       * defaults to true.
       */
      public void setUseIterator(boolean useIterator) {
          this.useIterator = useIterator;
      }
  
      /**
       * Indicates whether empty resultset should be allowed to be sent to the next hop or not
       */
      public boolean isRouteEmptyResultSet() {
          return routeEmptyResultSet;
      }
  
      /**
       * Sets whether empty resultset should be allowed to be sent to the next hop. defaults to false. So the empty
       * resultset will be filtered out.
       */
      public void setRouteEmptyResultSet(boolean routeEmptyResultSet) {
          this.routeEmptyResultSet = routeEmptyResultSet;
      }
  }
  