package org.rabix.engine.processor.handler.impl;

import org.rabix.bindings.model.dag.DAGContainer;
import org.rabix.bindings.model.dag.DAGLinkPort;
import org.rabix.bindings.model.dag.DAGLinkPort.LinkPortType;
import org.rabix.bindings.model.dag.DAGNode;
import org.rabix.engine.db.DAGNodeDB;
import org.rabix.engine.event.Event;
import org.rabix.engine.event.impl.InitEvent;
import org.rabix.engine.event.impl.InputUpdateEvent;
import org.rabix.engine.model.ContextRecord;
import org.rabix.engine.model.ContextRecord.ContextStatus;
import org.rabix.engine.model.JobRecord;
import org.rabix.engine.model.VariableRecord;
import org.rabix.engine.processor.EventProcessor;
import org.rabix.engine.processor.handler.EventHandler;
import org.rabix.engine.processor.handler.EventHandlerException;
import org.rabix.engine.service.ContextRecordService;
import org.rabix.engine.service.JobRecordService;
import org.rabix.engine.service.JobRecordService.JobState;
import org.rabix.engine.service.VariableRecordService;

import com.google.inject.Inject;

/**
 * Handles {@link InitEvent} events.
 */
public class InitEventHandler implements EventHandler<InitEvent> {

  private DAGNodeDB nodeDB;
  private EventProcessor eventProcessor;
  private JobRecordService jobRecordService;
  private ContextRecordService contextRecordService;
  private VariableRecordService variableRecordService;

  @Inject
  public InitEventHandler(EventProcessor eventProcessor, JobRecordService jobRecordService, VariableRecordService variableRecordService, ContextRecordService contextRecordService, DAGNodeDB dagNodeDB) {
    this.nodeDB = dagNodeDB;
    this.eventProcessor = eventProcessor;
    this.jobRecordService = jobRecordService;
    this.contextRecordService = contextRecordService;
    this.variableRecordService = variableRecordService;
  }

  public void handle(final InitEvent event) throws EventHandlerException {
    ContextRecord context = new ContextRecord(event.getContext().getId(), event.getContext().getConfig(), ContextStatus.RUNNING);
    
    contextRecordService.create(context);
    nodeDB.loadDB(event.getNode(), event.getContextId());
    
    DAGNode node = nodeDB.get(event.getNode().getId(), event.getContextId());
    JobRecord job = new JobRecord(event.getContextId(), event.getNode().getId(), event.getContextId(), JobState.PENDING, node instanceof DAGContainer, false, true);

    for (DAGLinkPort inputPort : node.getInputPorts()) {
      if (job.getState().equals(JobState.PENDING)) {
        job.incrementPortCounter(inputPort, LinkPortType.INPUT);
      }

      VariableRecord variable = new VariableRecord(event.getContextId(), event.getNode().getId(), inputPort.getId(), LinkPortType.INPUT, null);
      variableRecordService.create(variable);
    }

    for (DAGLinkPort outputPort : node.getOutputPorts()) {
      job.incrementPortCounter(outputPort, LinkPortType.OUTPUT);

      VariableRecord variable = new VariableRecord(event.getContextId(), event.getNode().getId(), outputPort.getId(), LinkPortType.OUTPUT, null);
      variableRecordService.create(variable);
    }
    jobRecordService.create(job);

    for (DAGLinkPort inputPort : node.getInputPorts()) {
      Object value = event.getValue().get(inputPort.getId());
      Event updateInputEvent = new InputUpdateEvent(event.getContextId(), event.getNode().getId(), inputPort.getId(), value, null);
      eventProcessor.send(updateInputEvent);
    }
  }

}

@@ -43,16 +43,93 @@
 .impl;%0A%0A
+import java.util.HashMap;%0Aimport java.util.Map;%0Aimport java.util.Map.Entry;%0A%0A
 import o
@@ -281,16 +281,60 @@
 rtType;%0A
+import org.rabix.common.helper.CloneHelper;%0A
 import o
@@ -3234,24 +3234,97 @@
 eate(job);%0A%0A
+    Map%3CString, Object%3E mixedInputs = mixInputs(node, event.getValue());%0A
     for (DAG
@@ -3389,32 +3389,27 @@
 value = 
-event.getValue()
+mixedInputs
 .get(inp
@@ -3611,12 +3611,638 @@
    %7D%0A  %7D
+%0A  %0A  @SuppressWarnings(%22unchecked%22)%0A  private Map%3CString, Object%3E mixInputs(DAGNode dagNode, Map%3CString, Object%3E inputs) %7B%0A    Map%3CString, Object%3E mixedInputs;%0A    try %7B%0A      mixedInputs = (Map%3CString, Object%3E) CloneHelper.deepCopy(dagNode.getDefaults());%0A      if (inputs == null) %7B%0A        return mixedInputs;%0A      %7D%0A%0A      for (Entry%3CString, Object%3E inputEntry : inputs.entrySet()) %7B%0A        mixedInputs.put(inputEntry.getKey(), inputEntry.getValue());%0A      %7D%0A      return mixedInputs;%0A    %7D catch (Exception e) %7B%0A      throw new RuntimeException(%22Failed to clone default inputs for node %22 + dagNode.getId());%0A    %7D%0A  %7D
 %0A%0A%7D%0A

add default values for root job