/*
 * Copyright 2020 Netflix, Inc.
 * <p>
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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 com.netflix.conductor.core.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.netflix.conductor.core.events.ActionProcessor;
import com.netflix.conductor.core.events.EventProcessor;
import com.netflix.conductor.core.events.EventQueueProvider;
import com.netflix.conductor.core.events.EventQueues;
import com.netflix.conductor.core.events.SimpleActionProcessor;
import com.netflix.conductor.core.events.SimpleEventProcessor;
import com.netflix.conductor.core.events.queue.ConductorEventQueueProvider;
import com.netflix.conductor.core.execution.WorkflowExecutor;
import com.netflix.conductor.core.utils.JsonUtils;
import com.netflix.conductor.core.utils.ParametersUtils;
import com.netflix.conductor.dao.QueueDAO;
import com.netflix.conductor.service.ExecutionService;
import com.netflix.conductor.service.MetadataService;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import rx.Scheduler;
import rx.schedulers.Schedulers;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

@Configuration(proxyBeanMethods = false)
public class EventConfiguration {

    @Bean
    public ActionProcessor actionProcessor(WorkflowExecutor workflowExecutor, ParametersUtils parametersUtils,
        JsonUtils jsonUtils) {
        return new SimpleActionProcessor(workflowExecutor, parametersUtils, jsonUtils);
    }

    @Bean
    public EventProcessor eventProcessor(ExecutionService executionService, MetadataService metadataService,
        ActionProcessor actionProcessor, EventQueues eventQueues, JsonUtils jsonUtils, ConductorProperties properties,
        ObjectMapper objectMapper) {
        return new SimpleEventProcessor(executionService, metadataService, actionProcessor, eventQueues, jsonUtils,
            properties, objectMapper);
    }

    @Bean
    public Scheduler scheduler(ConductorProperties properties) {
        ThreadFactory threadFactory = new ThreadFactoryBuilder()
            .setNameFormat("event-queue-poll-scheduler-thread-%d")
            .build();
        Executor executorService = Executors
            .newFixedThreadPool(properties.getEventSchedulerPollThreadCount(), threadFactory);

        return Schedulers.from(executorService);
    }

    @Bean
    public EventQueueProvider conductorEventQueueProvider(QueueDAO queueDAO, ConductorProperties properties,
        Scheduler scheduler) {
        return new ConductorEventQueueProvider(queueDAO, properties, scheduler);
    }
}

@@ -1499,24 +1499,103 @@
 ataService;%0A
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;%0A
 import org.s
@@ -1875,46 +1875,354 @@
 y;%0A%0A
-@Configuration(proxyBeanMethods = fals
+/**%0A * Default event processing configuration that is enabled by default.%0A *%0A * %3Cp%3ESet %3Ccode%3Eworkflow.event.processing.enabled=false%3C/code%3E to %3Ci%3Edisable%3C/i%3E event processing.%3C/p%3E%0A */%0A@Configuration(proxyBeanMethods = false)%0A@ConditionalOnProperty(prefix = %22workflow%22, name = %22event.processing.enabled%22, havingValue = %22true%22, matchIfMissing = tru
 e)%0Ap
@@ -2372,24 +2372,59 @@
 etersUtils,%0A
+                                   
         Json
@@ -2656,16 +2656,49 @@
 ervice,%0A
+                                 
         
@@ -2800,32 +2800,65 @@
 ies properties,%0A
+                                 
         ObjectMa
@@ -3006,16 +3006,20 @@
         
+    
 properti
@@ -3196,16 +3196,20 @@
         
+    
 .setName
@@ -3255,16 +3255,20 @@
 ad-%25d%22)%0A
+    
         
@@ -3326,16 +3326,20 @@
 ecutors%0A
+    
         
@@ -3478,37 +3478,445 @@
 ce);%0A    %7D%0A%0A    
-@Bean
+/**%0A     * Default provider for %7B@link com.netflix.conductor.core.events.queue.ObservableQueue%7D%0A     * that listens on the %3Ci%3Econductor%3C/i%3E queue prefix.%0A     *%0A     * %3Cp%3E%3Ccode%3ESet workflow.default.event.queue.enabled=false%3C/code%3E to disable the default queue.%3C/p%3E%0A     */%0A    @Bean%0A    @ConditionalOnProperty(prefix = %22workflow%22, name = %22default.event.queue.enabled%22, havingValue = %22true%22, matchIfMissing = true)
 %0A    public Even
@@ -4001,32 +4001,82 @@
 ies properties,%0A
+                                                  
         Schedule

Introduce properties that control event processing and event queues.