package reactor.spring.context.config;

import java.util.Map;
import java.util.function.Supplier;

import org.reactivestreams.Processor;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Computations;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Timer;
import reactor.core.util.PlatformDependent;
import reactor.spring.factory.CreateOrReuseFactoryBean;

import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
import org.springframework.core.type.AnnotationMetadata;
import org.springframework.util.StringUtils;

/**
 * {@link ImportBeanDefinitionRegistrar} implementation that configures necessary Reactor components.
 *
 * @author Jon Brisbin
 * @author Stephane Maldini
 */
public class ReactorBeanDefinitionRegistrar implements ImportBeanDefinitionRegistrar {

	private static final String DEFAULT_TIMER_SUPPLIER_NAME  = "reactorTimer";
	private static final String DEFAULT_SCHEDULER_GROUP_NAME = "reactorGroupedProcessors";

	private static final Supplier<Supplier<Processor>> DEFAULT_SCHEDULER_GROUP = new
			Supplier<Supplier<Processor>>() {
		@Override
		public Supplier<Processor> get() {
			final Scheduler group =
					Computations.parallel(DEFAULT_SCHEDULER_GROUP_NAME + "-spring", PlatformDependent
							.MEDIUM_BUFFER_SIZE);

			return new Supplier<Processor>() {
				@Override
				public Processor get() {
					return EmitterProcessor.async(group);
				}
			};

		}
	};

	private static final Supplier<Supplier<Timer>> DEFAULT_TIMER_SUPPLIER =  new Supplier<Supplier<Timer>>() {
		@Override
		public Supplier<Timer> get() {
			final Timer timer = Timer.create();
			return new Supplier<Timer>() {
				@Override
				public Timer get() {
					return timer;
				}
			};
		}
	};

	protected <T> void registerReactorBean(BeanDefinitionRegistry registry,
	                                       String attrValue,
	                                       String name, Class<T> tClass, Supplier<Supplier<T>> supplier) {

		// Create a root Enivronment
		if (!registry.containsBeanDefinition(name)) {
			BeanDefinitionBuilder envBeanDef = BeanDefinitionBuilder.rootBeanDefinition(CreateOrReuseFactoryBean
			  .class);
			envBeanDef.addConstructorArgValue(name);
			envBeanDef.addConstructorArgValue(tClass);

			if (StringUtils.hasText(attrValue)) {
				envBeanDef.addConstructorArgReference(attrValue);
			} else {
				envBeanDef.addConstructorArgValue(supplier.get());
			}
			registry.registerBeanDefinition(name, envBeanDef.getBeanDefinition());
		}
	}

	@Override
	public void registerBeanDefinitions(AnnotationMetadata meta, BeanDefinitionRegistry registry) {
		Map<String, Object> attrs = meta.getAnnotationAttributes(EnableReactor.class.getName());

		registerReactorBean(registry,
		  (String) attrs.get("timerSupplier"),
		  DEFAULT_TIMER_SUPPLIER_NAME,
		  Timer.class,
		  DEFAULT_TIMER_SUPPLIER
		);


		registerReactorBean(registry,
				(String) attrs.get("processorSupplier"),
				DEFAULT_SCHEDULER_GROUP_NAME,
				Processor.class,
				DEFAULT_SCHEDULER_GROUP
		);


		// Create a ConsumerBeanAutoConfiguration
		if (!registry.containsBeanDefinition(ConsumerBeanAutoConfiguration.class.getName())) {
			BeanDefinitionBuilder autoConfigDef = BeanDefinitionBuilder.rootBeanDefinition
			  (ConsumerBeanAutoConfiguration.class);
			registry.registerBeanDefinition(ConsumerBeanAutoConfiguration.class.getName(), autoConfigDef
			  .getBeanDefinition());
		}
	}
}

@@ -122,32 +122,76 @@
 eams.Processor;%0A
+import reactor.core.publisher.Computations;%0A
 import reactor.c
@@ -248,36 +248,37 @@
 e.publisher.
-Computations
+FluxProcessor
 ;%0Aimport rea
@@ -1270,99 +1270,16 @@
 P = 
-new%0A%09%09%09Supplier%3CSupplier%3CProcessor%3E%3E() %7B%0A%09%09@Override%0A%09%09public Supplier%3CProcessor%3E get()
+() -%3E
  %7B%0A
-%09
 %09%09fi
@@ -1300,17 +1300,16 @@
 group =%0A
-%09
 %09%09%09%09Comp
@@ -1392,17 +1392,16 @@
 t%0A%09%09%09%09%09%09
-%09
 .MEDIUM_
@@ -1415,17 +1415,19 @@
 SIZE);%0A%0A
-%09
+%0A%0A%0A
 %09%09return
@@ -1419,36 +1419,33 @@
 );%0A%0A%0A%0A%0A%09%09return 
-new 
+(
 Supplier%3CProcess
@@ -1451,58 +1451,72 @@
 sor%3E
-() %7B%0A%09%09%09%09@Override%0A%09%09%09%09public Processor get() %7B%0A%09%09
+) () -%3E %7B%0A%09%09%09FluxProcessor emitter = EmitterProcessor.create();%0A
 %09%09%09r
@@ -1521,23 +1521,20 @@
 %09return 
-Emitter
+Flux
 Processo
@@ -1539,38 +1539,54 @@
 sor.
-async
+wrap(emitter, emitter.publishOn
 (group)
+)
 ;%0A%09%09
-%09%09%7D%0A%09%09%09%7D;%0A%0A%09%09%7D
+%7D;%0A
 %0A%09%7D;
@@ -1664,89 +1664,16 @@
 R = 
- new Supplier%3CSupplier%3CTimer%3E%3E() %7B%0A%09%09@Override%0A%09%09public Supplier%3CTimer%3E get()
+() -%3E
  %7B%0A
-%09
 %09%09fi
@@ -1712,113 +1712,27 @@
 ;%0A%09%09
-%09
 return 
-new Supplier%3CTimer%3E() %7B%0A%09%09%09%09@Override%0A%09%09%09%09public Timer get() %7B%0A%09%09%09%09%09return timer;%0A%09%09%09%09%7D%0A%09%09%09%7D;%0A%09%09%7D
+() -%3E timer;
 %0A%09%7D;

Split FluxProcessor factories into individual Processor, remove confusing factories