/*
 * 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.uima.aae.handler;

import org.apache.uima.UIMAFramework;
import org.apache.uima.aae.UIMAEE_Constants;
import org.apache.uima.aae.UimaSerializer;
import org.apache.uima.aae.InProcessCache.CacheEntry;
import org.apache.uima.aae.controller.AggregateAnalysisEngineController;
import org.apache.uima.aae.controller.AnalysisEngineController;
import org.apache.uima.aae.controller.Endpoint;
import org.apache.uima.aae.controller.PrimitiveAnalysisEngineController;
import org.apache.uima.aae.controller.LocalCache.CasStateEntry;
import org.apache.uima.aae.error.AsynchAEException;
import org.apache.uima.aae.error.ErrorContext;
import org.apache.uima.aae.jmx.ServicePerformance;
import org.apache.uima.aae.message.AsynchAEMessage;
import org.apache.uima.aae.message.MessageContext;
import org.apache.uima.aae.message.UIMAMessage;
import org.apache.uima.aae.monitor.Monitor;
import org.apache.uima.cas.CAS;
import org.apache.uima.util.Level;

public abstract class HandlerBase implements Handler {
  private static final Class CLASS_NAME = HandlerBase.class;

  private Handler delegateHandler;

  private Handler parentHandler;

  private String handlerName;

  private AnalysisEngineController controller;

  public AnalysisEngineController getController() {
    return controller;
  }

  public void setController(AnalysisEngineController controller) {
    this.controller = controller;
  }

  public static ErrorContext populateErrorContext(MessageContext aMessageCtx) {
    ErrorContext errorContext = new ErrorContext();
    if (aMessageCtx != null) {
      try {
        if (aMessageCtx.propertyExists(AsynchAEMessage.Command)) {
          errorContext.add(AsynchAEMessage.Command, aMessageCtx
                  .getMessageIntProperty(AsynchAEMessage.Command));
        }

        if (aMessageCtx.propertyExists(AsynchAEMessage.MessageType)) {
          errorContext.add(AsynchAEMessage.MessageType, aMessageCtx
                  .getMessageIntProperty(AsynchAEMessage.MessageType));
        }

        if (aMessageCtx.propertyExists(AsynchAEMessage.CasReference)) {
          errorContext.add(AsynchAEMessage.CasReference, aMessageCtx
                  .getMessageStringProperty(AsynchAEMessage.CasReference));
        }
        errorContext.add(UIMAMessage.RawMsg, aMessageCtx.getRawMessage());
      } catch (Exception e) { /* ignore */
      }
    }
    return errorContext;
  }

  public void validate(Object anObjectToHandle) throws AsynchAEException {
    if (anObjectToHandle == null) {
      throw new AsynchAEException("Nothing to Handle - Input Object is Null");
    }
    if (!(anObjectToHandle instanceof MessageContext)) {
      throw new AsynchAEException(
              "Invalid Object in Handler. Expected MessageContext Instead Got::"
                      + anObjectToHandle.getClass().getName());
    }
  }

  public boolean isHandlerForMessage(MessageContext aMessageContext, int anExpectedMessageType,
          int anExpectedCommand) throws AsynchAEException {
    int messageType = aMessageContext.getMessageIntProperty(AsynchAEMessage.MessageType);
    int command = aMessageContext.getMessageIntProperty(AsynchAEMessage.Command);
    if (anExpectedMessageType == messageType && anExpectedCommand == command) {
      return true;
    }
    return false;
  }

  public void invokeProcess(CAS aCAS, String anInputCasReferenceId, String aNewCasReferenceId,
          MessageContext aMessageContext, String aNewCasProducedBy) throws AsynchAEException {
    try {
      // Use empty string as key. Top level component stats are stored under this key.
      controller.getMonitor().incrementCount("", Monitor.ProcessCount);

      if (controller instanceof AggregateAnalysisEngineController) {
        boolean isNewCAS = aMessageContext.propertyExists(AsynchAEMessage.CasSequence);
        if (isNewCAS) {
          ((AggregateAnalysisEngineController) controller).process(aCAS, anInputCasReferenceId,
                  aNewCasReferenceId, aNewCasProducedBy);
        } else {
          ((AggregateAnalysisEngineController) controller).process(aCAS, anInputCasReferenceId);
        }
      } else if (controller instanceof PrimitiveAnalysisEngineController) {
        ((PrimitiveAnalysisEngineController) controller).process(aCAS, anInputCasReferenceId,
                aMessageContext.getEndpoint());
      } else {
        throw new AsynchAEException(
                "Invalid Controller. Expected AggregateController or PrimitiveController. Got:"
                        + controller.getClass().getName());
      }
    } catch (AsynchAEException e) {
      throw e;
    } catch (Exception e) {
      throw new AsynchAEException(e);
    }

  }

  public void delegate(MessageContext aMessageContext) throws AsynchAEException {
    int messageType = aMessageContext.getMessageIntProperty(AsynchAEMessage.MessageType);
    if (hasDelegateHandler()) {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(
                Level.FINE,
                CLASS_NAME.getName(),
                "delegate",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_dispatch_msg_to_handler__FINE",
                new Object[] { messageType, aMessageContext.getEndpoint().getEndpoint(),
                    getDelegate().getName() });
      }
      getDelegate().handle(aMessageContext);
    } else {
      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "delegate",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_no_handler_for_message__FINE",
                new Object[] { messageType, aMessageContext.getEndpoint().getEndpoint() });
      }
    }
  }

  public HandlerBase(String aName) {
    handlerName = aName;
  }

  public Handler getDelegate() {
    return delegateHandler;
  }

  public void setParent(Handler aHandler) {
    parentHandler = aHandler;
  }

  public String getName() {
    return handlerName;
  }

  public void setDelegate(Handler aHandler) {
    delegateHandler = aHandler;
  }

  public boolean hasDelegateHandler() {
    return (delegateHandler != null);
  }

  public void removeDelegate(String aHandlerName) {
    if (handlerName.equals(aHandlerName)) {
      if (hasDelegateHandler()) {
        // connect the delegate with the parent of this handler
        ((HandlerBase) delegateHandler).setParent(parentHandler);
        parentHandler.setDelegate(delegateHandler);
      }
    } else {
      delegateHandler.removeDelegate(aHandlerName);
    }

  }

  public void resequenceDelegateHandler(String aHandlerName, int aNewPositionInHandlerChain) {
    // TODO Auto-generated method stub

  }

  public void handle(Object anObjectToHandle) throws AsynchAEException {
    getController().getControllerLatch().waitUntilInitialized();
  }

  public void handle(Object anObjectToHandle, String expectedOutputType) throws AsynchAEException {
  }

  protected synchronized void aggregateDelegateStats(MessageContext aMessageContext,
          String aCasReferenceId) throws AsynchAEException {
    String delegateKey = "";
    try {

      delegateKey = ((AggregateAnalysisEngineController) getController())
              .lookUpDelegateKey(aMessageContext.getEndpoint().getEndpoint());
      CacheEntry entry = getController().getInProcessCache().getCacheEntryForCAS(aCasReferenceId);
      if (entry == null) {
        throw new AsynchAEException("CasReferenceId:" + aCasReferenceId
                + " Not Found in the Cache.");
      }
      CacheEntry inputCasEntry = null;
      String inputCasReferenceId = entry.getInputCasReferenceId();
      ServicePerformance casStats = ((AggregateAnalysisEngineController) getController())
              .getCasStatistics(aCasReferenceId);
      if (inputCasReferenceId != null
              && getController().getInProcessCache().entryExists(inputCasReferenceId)) {
        String casProducerKey = entry.getCasProducerKey();
        if (casProducerKey != null
                && ((AggregateAnalysisEngineController) getController())
                        .isDelegateKeyValid(casProducerKey)) {
          // Get entry for the input CAS
          inputCasEntry = getController().getInProcessCache().getCacheEntryForCAS(
                  inputCasReferenceId);
        }

      }
      ServicePerformance delegateServicePerformance = ((AggregateAnalysisEngineController) getController())
              .getServicePerformance(delegateKey);

      if (aMessageContext.propertyExists(AsynchAEMessage.TimeToSerializeCAS)) {
        long timeToSerializeCAS = ((Long) aMessageContext
                .getMessageLongProperty(AsynchAEMessage.TimeToSerializeCAS)).longValue();
        if (timeToSerializeCAS > 0) {
          if (delegateServicePerformance != null) {
            delegateServicePerformance.incrementCasSerializationTime(timeToSerializeCAS);
          }
        }
      }
      if (aMessageContext.propertyExists(AsynchAEMessage.TimeToDeserializeCAS)) {
        long timeToDeserializeCAS = ((Long) aMessageContext
                .getMessageLongProperty(AsynchAEMessage.TimeToDeserializeCAS)).longValue();
        if (timeToDeserializeCAS > 0) {
          if (delegateServicePerformance != null) {
            delegateServicePerformance.incrementCasDeserializationTime(timeToDeserializeCAS);
          }
        }
      }

      if (aMessageContext.propertyExists(AsynchAEMessage.IdleTime)) {
        long idleTime = ((Long) aMessageContext.getMessageLongProperty(AsynchAEMessage.IdleTime))
                .longValue();
        if (idleTime > 0 && delegateServicePerformance != null) {
          Endpoint endp = aMessageContext.getEndpoint();
          if (endp != null && endp.isRemote()) {
            delegateServicePerformance.incrementIdleTime(idleTime);
          }
        }
      }

      if (aMessageContext.propertyExists(AsynchAEMessage.TimeWaitingForCAS)) {
        long timeWaitingForCAS = ((Long) aMessageContext
                .getMessageLongProperty(AsynchAEMessage.TimeWaitingForCAS)).longValue();
        if (timeWaitingForCAS > 0 && aMessageContext.getEndpoint().isRemote()) {
          entry.incrementTimeWaitingForCAS(timeWaitingForCAS);
          delegateServicePerformance.incrementCasPoolWaitTime(timeWaitingForCAS
                  - delegateServicePerformance.getRawCasPoolWaitTime());
          if (inputCasEntry != null) {
            inputCasEntry.incrementTimeWaitingForCAS(timeWaitingForCAS);
          }
        }
      }
      if (aMessageContext.propertyExists(AsynchAEMessage.TimeInProcessCAS)) {
        long timeInProcessCAS = ((Long) aMessageContext
                .getMessageLongProperty(AsynchAEMessage.TimeInProcessCAS)).longValue();
        Endpoint endp = aMessageContext.getEndpoint();
        if (endp != null && endp.isRemote()) {
          if (delegateServicePerformance != null) {
            // calculate the time spent in analysis. The remote service returns total time
            // spent in the analysis. Compute the delta.
            long dt = timeInProcessCAS - delegateServicePerformance.getRawAnalysisTime();
            // increment total time in analysis
            delegateServicePerformance.incrementAnalysisTime(dt);
            getController().getServicePerformance().incrementAnalysisTime(dt);
          }
        } else {
          getController().getServicePerformance().incrementAnalysisTime(timeInProcessCAS);
        }
        casStats.incrementAnalysisTime(timeInProcessCAS);

        if (inputCasReferenceId != null) {
          ServicePerformance inputCasStats = ((AggregateAnalysisEngineController) getController())
                  .getCasStatistics(inputCasReferenceId);
          // Update processing time for this CAS
          if (inputCasStats != null) {
            inputCasStats.incrementAnalysisTime(timeInProcessCAS);
          }
        }

      }
    } catch (AsynchAEException e) {
      throw e;
    } catch (Exception e) {
      throw new AsynchAEException(e);
    }
  }

  protected void computeStats(MessageContext aMessageContext, String aCasReferenceId)
          throws AsynchAEException {
    if (aMessageContext.propertyExists(AsynchAEMessage.TimeInService)) {
      long departureTime = getController().getTime(aCasReferenceId,
              aMessageContext.getEndpoint().getEndpoint());
      long currentTime = System.nanoTime();
      long roundTrip = currentTime - departureTime;
      long timeInService = aMessageContext.getMessageLongProperty(AsynchAEMessage.TimeInService);
      long totalTimeInComms = currentTime - (departureTime - timeInService);

      if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
        UIMAFramework.getLogger(CLASS_NAME).logrb(
                Level.FINE,
                CLASS_NAME.getName(),
                "computeStats",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_show_roundtrip_time__FINE",
                new Object[] { aCasReferenceId, aMessageContext.getEndpoint(),
                    (double) roundTrip / (double) 1000000 });

        UIMAFramework.getLogger(CLASS_NAME).logrb(
                Level.FINE,
                CLASS_NAME.getName(),
                "computeStats",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_show_time_spent_in_delegate__FINE",
                new Object[] { aCasReferenceId, (double) timeInService / (double) 1000000,
                    aMessageContext.getEndpoint() });

        UIMAFramework.getLogger(CLASS_NAME).logrb(
                Level.FINE,
                CLASS_NAME.getName(),
                "computeStats",
                UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE,
                "UIMAEE_show_time_spent_in_comms__FINE",
                new Object[] { aCasReferenceId, (double) totalTimeInComms / (double) 1000000,
                    aMessageContext.getEndpoint() });
      }
    }

    if (getController() instanceof AggregateAnalysisEngineController) {
      aggregateDelegateStats(aMessageContext, aCasReferenceId);
    }
  }

}