package reactor.spring.context.config;

import java.util.Map;

import org.reactivestreams.Processor;
import reactor.core.publisher.ProcessorGroup;
import reactor.core.timer.Timer;
import reactor.core.util.PlatformDependent;
import reactor.fn.Supplier;
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_PROCESSOR_SUPPLIER_NAME = "reactorProcessor";

	private static final Supplier<Supplier<? extends Processor>> DEFAULT_PROCESSOR_SUPPLIER = new Supplier<Supplier<?
	  extends Processor>>() {
		@Override
		public Supplier<? extends Processor> get() {
			return ProcessorGroup.async("reactor-spring",
			  PlatformDependent.MEDIUM_BUFFER_SIZE
			);
		}
	};

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

	protected <T> void registerReactorBean(BeanDefinitionRegistry registry,
	                                       String attrValue,
	                                       String name,
	                                       Class<T> tClass,
	                                       Supplier<Supplier<? extends 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_PROCESSOR_SUPPLIER_NAME,
		  Processor.class,
		  DEFAULT_PROCESSOR_SUPPLIER
		);


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

@@ -60,46 +60,8 @@
 p;%0A%0A
-import org.reactivestreams.Processor;%0A
 impo
@@ -86,24 +86,24 @@
 blisher.
-Processo
+Schedule
 rGroup;%0A
@@ -899,19 +899,16 @@
 R_NAME  
-   
 = %22react
@@ -954,34 +954,31 @@
 DEFAULT_
-PROCESSOR_SUPPLIER
+SCHEDULER_GROUP
 _NAME = 
@@ -985,25 +985,30 @@
 %22reactor
-Processor
+SchedulerGroup
 %22;%0A%0A%09pri
@@ -1039,168 +1039,117 @@
 er%3CS
-upplier%3C? extends Processor%3E%3E DEFAULT_PROCESSOR_SUPPLIER = new Supplier%3CSupplier%3C?%0A%09  extends Processor%3E%3E() %7B%0A%09%09@Override%0A%09%09public Supplier%3C? extends Processor%3E
+chedulerGroup%3E DEFAULT_SCHEDULER_GROUP = new Supplier%3CSchedulerGroup%3E() %7B%0A%09%09@Override%0A%09%09public SchedulerGroup
  get
@@ -1163,24 +1163,24 @@
 %09return 
-Processo
+Schedule
 rGroup.a
@@ -1184,24 +1184,48 @@
 p.async(
-%22reactor
+DEFAULT_SCHEDULER_GROUP_NAME + %22
 -spring%22
@@ -1225,21 +1225,16 @@
 spring%22,
-%0A%09%09%09 
  Platfor
@@ -1262,20 +1262,16 @@
 FER_SIZE
-%0A%09%09%09
 );%0A%09%09%7D%0A%09
@@ -1309,34 +1309,14 @@
 ier%3C
-Supplier%3C? extends 
 Timer%3E
-%3E
  DEF
@@ -1340,120 +1340,8 @@
 ER =
- new Supplier%3CSupplier%3C?%0A%09  extends Timer%3E%3E() %7B%0A%09%09@Override%0A%09%09public Supplier%3C? extends Timer%3E get() %7B%0A%09%09%09return
  new
@@ -1438,18 +1438,8 @@
 %09%09%7D%0A
-%09%09%09%7D;%0A%09%09%7D%0A
 %09%7D;%0A
@@ -1626,115 +1626,26 @@
 ame,
-%0A%09                                       Class%3CT%3E tClass,%0A%09                                       Supplier%3C
+ Class%3CT%3E tClass, 
 Supp
@@ -1661,17 +1661,16 @@
 tends T%3E
-%3E
  supplie
@@ -2595,34 +2595,34 @@
 ean(registry,%0A%09%09
-  
+%09%09
 (String) attrs.g
@@ -2644,34 +2644,34 @@
 pplier%22),%0A%09%09
-  
+%09%09
 DEFAULT_
 PROCESSOR_SU
@@ -2662,46 +2662,48 @@
 ULT_
-PROCESSOR_SUPPLIER_NAME,%0A%09%09  Processor
+SCHEDULER_GROUP_NAME,%0A%09%09%09%09SchedulerGroup
 .cla
@@ -2708,26 +2708,26 @@
 lass,%0A%09%09
-  
+%09%09
 DEFAULT_
 PROCESSO
@@ -2722,26 +2722,23 @@
 ULT_
-PROCESSOR_SUPPLIER
+SCHEDULER_GROUP
 %0A%09%09)

rename ProcessorGroup to SchedulerGroup, Simplify SchedulerGroup and align with RSC scheduling conventions +Mono#defer +FluxProcessor#blocking +FluxProcessor#async +Broadcaster#blocking +Broadcaster#async