package org.rabix.engine.processor.impl;

import java.util.Set;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import org.rabix.bindings.model.Job;
import org.rabix.engine.event.Event;
import org.rabix.engine.event.Event.EventStatus;
import org.rabix.engine.event.Event.EventType;
import org.rabix.engine.event.impl.ContextStatusEvent;
import org.rabix.engine.model.ContextRecord;
import org.rabix.engine.model.ContextRecord.ContextStatus;
import org.rabix.engine.processor.EventProcessor;
import org.rabix.engine.processor.handler.EventHandlerException;
import org.rabix.engine.processor.handler.HandlerFactory;
import org.rabix.engine.repository.EventRepository;
import org.rabix.engine.repository.JobRepository;
import org.rabix.engine.repository.TransactionHelper;
import org.rabix.engine.repository.TransactionHelper.TransactionException;
import org.rabix.engine.service.CacheService;
import org.rabix.engine.service.ContextRecordService;
import org.rabix.engine.service.JobService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.inject.Inject;

/**
 * Event processor implementation
 */
public class EventProcessorImpl implements EventProcessor {

  private static final Logger logger = LoggerFactory.getLogger(EventProcessorImpl.class);
  
  public final static long SLEEP = 100;
  
  private final BlockingQueue<Event> events = new LinkedBlockingQueue<>();
  private final BlockingQueue<Event> externalEvents = new LinkedBlockingQueue<>();
  
  private final ExecutorService executorService = Executors.newSingleThreadExecutor();

  private final AtomicBoolean stop = new AtomicBoolean(false);
  private final AtomicBoolean running = new AtomicBoolean(false);

  private final HandlerFactory handlerFactory;
  
  private final ContextRecordService contextRecordService;
  
  private final TransactionHelper transactionHelper;
  private final CacheService cacheService;
  
  private final JobRepository jobRepository;
  private final EventRepository eventRepository;
  private final JobService jobService;
  
  @Inject
  public EventProcessorImpl(HandlerFactory handlerFactory, ContextRecordService contextRecordService,
      TransactionHelper transactionHelper, CacheService cacheService, EventRepository eventRepository,
      JobRepository jobRepository, JobService jobService) {
    this.handlerFactory = handlerFactory;
    this.contextRecordService = contextRecordService;
    this.transactionHelper = transactionHelper;
    this.cacheService = cacheService;
    this.eventRepository = eventRepository;
    this.jobRepository = jobRepository;
    this.jobService = jobService;
  }

  public void start() {
    executorService.execute(new Runnable() {
      @Override
      public void run() {
        final AtomicReference<Event> eventReference = new AtomicReference<Event>(null);
        while (!stop.get()) {
          try {
            eventReference.set(externalEvents.poll());
            if (eventReference.get() == null) {
              running.set(false);
              Thread.sleep(SLEEP);
              continue;
            }
            running.set(true);
            transactionHelper.doInTransaction(new TransactionHelper.TransactionCallback<Void>() {
              @Override
              public Void call() throws TransactionException {
                handle(eventReference.get());
                cacheService.flush(eventReference.get().getContextId());
                
                Set<Job> readyJobs = jobRepository.getReadyJobsByGroupId(eventReference.get().getEventGroupId());
                jobService.handleJobsReady(readyJobs, eventReference.get().getContextId(), eventReference.get().getProducedByNode());
 
//                eventRepository.update(eventReference.get().getEventGroupId(), eventReference.get().getPersistentType(), Event.EventStatus.PROCESSED);
                eventRepository.delete(eventReference.get().getEventGroupId());
                return null;
              }
            });
          } catch (Exception e) {
            logger.error("EventProcessor failed to process event {}.", eventReference.get(), e);
            try {
              invalidateContext(eventReference.get().getContextId());
            } catch (EventHandlerException ehe) {
              logger.error("Failed to invalidate Context {}.", eventReference.get().getContextId(), ehe);
              stop();
            }
          }
        }
      }
    });
  }
  
  private void handle(Event event) throws TransactionException {
    while (event != null) {
      try {
        ContextRecord context = contextRecordService.find(event.getContextId());
        if (context != null && (context.getStatus().equals(ContextStatus.FAILED) || context.getStatus().equals(ContextStatus.ABORTED))) {
          logger.info("Skip event {}. Context {} has been invalidated.", event, context.getId());
          return;
        }
        handlerFactory.get(event.getType()).handle(event);
      } catch (EventHandlerException e) {
        throw new TransactionException(e);
      }
      event = events.poll();
    }
  }
  
  /**
   * Invalidates context 
   */
  private void invalidateContext(UUID contextId) throws EventHandlerException {
    handlerFactory.get(Event.EventType.CONTEXT_STATUS_UPDATE).handle(new ContextStatusEvent(contextId, ContextStatus.FAILED));
  }
  
  @Override
  public void stop() {
    stop.set(true);
    running.set(false);
  }

  public boolean isRunning() {
    return running.get();
  }

  public void send(Event event) throws EventHandlerException {
    if (stop.get()) {
      return;
    }
    if (event.getType().equals(EventType.INIT)) {
      addToQueue(event);
      return;
    }
    handlerFactory.get(event.getType()).handle(event);
  }

  public void addToQueue(Event event) {
    if (stop.get()) {
      return;
    }
    this.events.add(event);
  }
  
  @Override
  public void persist(Event event) {
    if (stop.get()) {
      return;
    }
    eventRepository.insert(event.getEventGroupId(), event.getPersistentType(), event, EventStatus.UNPROCESSED);    
  }
  
  public void addToExternalQueue(Event event) {
    if (stop.get()) {
      return;
    }
    this.externalEvents.add(event);
  }

}

@@ -529,16 +529,73 @@
 ntType;%0A
+import org.rabix.engine.event.Event.PersistentEventType;%0A
 import o
@@ -3737,16 +3737,81 @@
        %0A
+                if (checkForReadyJobs(eventReference.get())) %7B%0A  
         
@@ -3908,32 +3908,34 @@
 ventGroupId());%0A
+  
                 
@@ -4051,18 +4051,36 @@
 Node());
-%0A
  
+ %0A                %7D
 %0A//     
@@ -4818,24 +4818,369 @@
  %7D);%0A  %7D%0A  %0A
+  private boolean checkForReadyJobs(Event event) %7B%0A    switch (event.getType()) %7B%0A    case INIT:%0A      return true;%0A    case JOB_STATUS_UPDATE:%0A      if (PersistentEventType.JOB_STATUS_UPDATE_COMPLETED.equals(event.getPersistentType())) %7B%0A        return true;%0A      %7D%0A      return false;%0A    default:%0A      break;%0A    %7D%0A    return false;%0A  %7D%0A  %0A
   private vo

optimize event processors