/*
    * Copyright 2008 the original author or authors.
    * 
    * Licensed 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.openehealth.ipf.platform.camel.core.adapter;
  
  import groovy.lang.Closure;
  import groovy.transform.stc.ClosureParams;
  import groovy.transform.stc.SimpleType;
  import org.apache.camel.Exchange;
  import org.apache.camel.Expression;
  import org.apache.camel.processor.aggregate.AggregationStrategy;
  import org.openehealth.ipf.commons.core.modules.api.Aggregator;
  import org.openehealth.ipf.platform.camel.core.closures.DelegatingExpression;
  
  import java.util.Arrays;
  
  import static org.apache.camel.builder.Builder.body;
  import static org.openehealth.ipf.platform.camel.core.util.Exchanges.prepareResult;
  
  /**
   * Adapts an {@link Aggregator}. 
   * 
   * @author Martin Krasser
   */
  public class AggregatorAdapter extends AdapterSupport implements AggregationStrategy {
  
      private Expression aggregationInputExpression;
  
      private final Aggregator aggregator;
  
      /**
       * Creates a new {@link AggregatorAdapter} and sets the delegate
       * {@link Aggregator}.
       * 
       * @param aggregator
       *            an aggregator.
       */
      public AggregatorAdapter(Aggregator aggregator) {
          this.aggregator = aggregator;
          aggregationInputExpression = body();
      }
      
      /**
       * Sets an {@link Expression} for obtaining data to be obtained from an
       * additional (new) {@link Exchange}. The default expression obtains the
       * body from the input message.
       * 
       * @param aggregationInputExpression
       *            expression for obtaining aggregation input data.
       * @return this object.
       * 
       * @see #aggregate(Exchange, Exchange)
       */
      public AggregatorAdapter aggregationInput(Expression aggregationInputExpression) {
          this.aggregationInputExpression = aggregationInputExpression;
          return this;
      }
  
      /**
       * Sets an expression {@link Closure} for obtaining data to be obtained from
       * an additional (new) {@link Exchange}. The default expression obtains the
       * body from the input message.
       * 
       * @param aggregationInputExpressionLogic
       *            expression for obtaining aggregation input data.
       * @return this object.
       * 
       * @see #aggregate(Exchange, Exchange)
       */
      public AggregatorAdapter aggregationInput(@ClosureParams(value = SimpleType.class, options = { "org.apache.camel.Expression"})
                                                        Closure aggregationInputExpressionLogic) {
          return aggregationInput(new DelegatingExpression(aggregationInputExpressionLogic));
      }
  
      /**
       * Applies expressions to <code>oldExchange</code> and
       * <code>newExchange</code> and delegates further processing to
       * {@link #doAggregate(Exchange, Object, Object, Object...)}
       * 
       * @see #aggregationInput(Expression)
       * @see #input(Expression)
       * @see #params(Expression)
       */
      @Override
      public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
          Object newInput = adaptAggregationInput(newExchange);
          Object oldInput = adaptInput(oldExchange);
         Object params = adaptParams(oldExchange);
         if (params == null) {
             doAggregate(oldExchange, oldInput, newInput, (Object[])null);
         } else if (params.getClass().isArray()) {
             doAggregate(oldExchange, oldInput, newInput, (Object[])params);
         } else {
             doAggregate(oldExchange, oldInput, newInput, params);
         }
         return oldExchange;
     }
 
     /**
      * Aggregates <code>oldInputData</code> and <code>newInputData</code>.
      * The aggregation result is written to body of the message returned by
      * {@link org.openehealth.ipf.platform.camel.core.util.Exchanges#resultMessage(Exchange)}.
      * 
      * @param oldExchange original message exchange to write results to.
      * @param oldInputData original input data
      * @param newInputData additional input data
      * @param inputParams input parameters
      */
     protected void doAggregate(Exchange oldExchange, Object oldInputData, 
             Object newInputData, Object... inputParams) {
         
         prepareResult(oldExchange).setBody(
                 aggregator.zap(Arrays.asList(oldInputData, newInputData), inputParams));
     }
     
     /**
      * Applies the {@link Expression} set by
      * {@link #aggregationInput(Expression)} to obtain input data from the
      * <code>exchange</code>.
      * 
      * @param exchange
      *            message exchange.
      * @return aggregation input data or <code>null</code> if the expression
      *         evaluates to <code>null</code> or the expression object is
      *         <code>null</code>.
      */
     private Object adaptAggregationInput(Exchange exchange) {
         if (aggregationInputExpression == null) {
             return null;
         }
         return aggregationInputExpression.evaluate(exchange, Object.class);
     }
 
 }