From 80408d2f77a90f9129e425542a62640a50c0a832 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Wed, 4 Jan 2017 09:05:56 +0000 Subject: [PATCH] Add support for consumers and factor out base class --- ...ntextFunctionCatalogAutoConfiguration.java | 192 ++++++++++-------- 1 file changed, 108 insertions(+), 84 deletions(-) diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfiguration.java index 4d8d8f1cd..dda20ffd1 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfiguration.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfiguration.java @@ -30,6 +30,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.BeanFactoryPostProcessor; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; @@ -91,6 +92,9 @@ public class ContextFunctionCatalogAutoConfiguration { else if (isGenericSupplier(factory, name)) { wrapSupplier(factory, name); } + else if (isGenericConsumer(factory, name)) { + wrapConsumer(factory, name); + } } } @@ -117,51 +121,77 @@ public class ContextFunctionCatalogAutoConfiguration { String.class))); } + private boolean isGenericConsumer(ConfigurableListableBeanFactory factory, + String name) { + return factory.isTypeMatch(name, + ResolvableType.forClassWithGenerics(Consumer.class, Flux.class)) + && !factory.isTypeMatch(name, + ResolvableType.forClassWithGenerics(Consumer.class, + ResolvableType.forClassWithGenerics(Flux.class, + String.class))); + } + private void wrapFunction(ConfigurableListableBeanFactory factory, String name) { - BeanDefinition definition = registry.getBeanDefinition(name); - BeanDefinitionBuilder builder = BeanDefinitionBuilder - .genericBeanDefinition(ProxyFunction.class); - builder.addPropertyValue("delegate", definition); - builder.addPropertyValue("name", name); - targets.registerBeanDefinition(name, definition); - builder.addPropertyValue("registry", targets); - registry.registerBeanDefinition(name, builder.getRawBeanDefinition()); + AbstractBeanDefinition wrapped = getInputAwareWrappedBean(name, + ProxyFunction.class).getBeanDefinition(); + registry.registerBeanDefinition(name, wrapped); } private void wrapSupplier(ConfigurableListableBeanFactory factory, String name) { + AbstractBeanDefinition wrapped = getWrappedBean(name, ProxySupplier.class) + .getBeanDefinition(); + registry.registerBeanDefinition(name, wrapped); + } + + private void wrapConsumer(ConfigurableListableBeanFactory factory, String name) { + AbstractBeanDefinition wrapped = getInputAwareWrappedBean(name, + ProxyConsumer.class).getBeanDefinition(); + registry.registerBeanDefinition(name, wrapped); + } + + private BeanDefinitionBuilder getInputAwareWrappedBean(String name, + Class type) { + BeanDefinitionBuilder builder = getWrappedBean(name, type); + builder.addPropertyValue("name", name); + targets.registerBeanDefinition(name, registry.getBeanDefinition(name)); + builder.addPropertyValue("registry", targets); + return builder; + } + + private BeanDefinitionBuilder getWrappedBean(String name, Class type) { BeanDefinition definition = registry.getBeanDefinition(name); BeanDefinitionBuilder builder = BeanDefinitionBuilder - .genericBeanDefinition(ProxySupplier.class); + .genericBeanDefinition(type); builder.addPropertyValue("delegate", definition); - builder.addPropertyValue("name", name); - targets.registerBeanDefinition(name, definition); - builder.addPropertyValue("registry", targets); - registry.registerBeanDefinition(name, builder.getRawBeanDefinition()); + return builder; } } } -class ProxyFunction implements Function, Flux> { +abstract class ProxyWrapper { private ObjectMapper mapper; - private Function, Flux> delegate; + private T delegate; private String name; - private Class type; + private Class type; private BeanDefinitionRegistry registry; - @Autowired - public ProxyFunction(ObjectMapper mapper) { + public ProxyWrapper(ObjectMapper mapper) { this.mapper = mapper; } - public void setDelegate(Function, Flux> delegate) { + public void setDelegate(T delegate) { this.delegate = delegate; } + public T getDelegate() { + return delegate; + } + public void setName(String name) { this.name = name; } @@ -170,92 +200,86 @@ class ProxyFunction implements Function, Flux> { this.registry = registry; } - private Class findType(RootBeanDefinition definition) { + private Class findType(RootBeanDefinition definition) { StandardMethodMetadata source = (StandardMethodMetadata) definition.getSource(); ParameterizedType type = (ParameterizedType) (source.getIntrospectedMethod() .getGenericReturnType()); type = (ParameterizedType) type.getActualTypeArguments()[0]; Type param = type.getActualTypeArguments()[0]; - return ClassUtils.resolveClassName(param.getTypeName(), + @SuppressWarnings("unchecked") + Class resolved = (Class) ClassUtils.resolveClassName(param.getTypeName(), registry.getClass().getClassLoader()); + return resolved; + } + + public Class getType() { + if (type == null) { + type = findType((RootBeanDefinition) registry.getBeanDefinition(name)); + } + return type; + } + + public S fromJson(String value) { + try { + return mapper.readValue(value, getType()); + } + catch (Exception e) { + throw new IllegalStateException("Cannot convert from JSON: " + value); + } + } + + public String toJson(Object value) { + try { + return mapper.writeValueAsString(value); + } + catch (Exception e) { + throw new IllegalStateException("Cannot convert to JSON: " + value); + } + } + +} + +class ProxyFunction extends ProxyWrapper, Flux>> + implements Function, Flux> { + + @Autowired + public ProxyFunction(ObjectMapper mapper) { + super(mapper); } @Override public Flux apply(Flux input) { - if (type == null) { - type = findType((RootBeanDefinition) registry.getBeanDefinition(name)); - } - return delegate.apply(input.map(value -> { - try { - return mapper.readValue(value, type); - } - catch (Exception e) { - throw new IllegalStateException("Cannot convert from JSON: " + input); - } - })).map(value -> { - try { - return mapper.writeValueAsString(value); - } - catch (Exception e) { - throw new IllegalStateException("Cannot convert to JSON: " + input); - } - }); + return getDelegate().apply(input.map(this::fromJson)).map(this::toJson); } } -class ProxySupplier implements Supplier> { - - private ObjectMapper mapper; - - private Supplier> delegate; - - private String name; - - private Class type; - - private BeanDefinitionRegistry registry; +class ProxySupplier extends ProxyWrapper>> + implements Supplier> { @Autowired public ProxySupplier(ObjectMapper mapper) { - this.mapper = mapper; - } - - public void setDelegate(Supplier> delegate) { - this.delegate = delegate; - } - - public void setName(String name) { - this.name = name; - } - - public void setRegistry(BeanDefinitionRegistry registry) { - this.registry = registry; - } - - private Class findType(RootBeanDefinition definition) { - StandardMethodMetadata source = (StandardMethodMetadata) definition.getSource(); - ParameterizedType type = (ParameterizedType) (source.getIntrospectedMethod() - .getGenericReturnType()); - type = (ParameterizedType) type.getActualTypeArguments()[0]; - Type param = type.getActualTypeArguments()[0]; - return ClassUtils.resolveClassName(param.getTypeName(), - registry.getClass().getClassLoader()); + super(mapper); } @Override public Flux get() { - if (type == null) { - type = findType((RootBeanDefinition) registry.getBeanDefinition(name)); - } - return delegate.get().map(value -> { - try { - return mapper.writeValueAsString(value); - } - catch (Exception e) { - throw new IllegalStateException("Cannot convert to JSON: " + value); - } - }); + return getDelegate().get().map(this::toJson); + } + +} + +class ProxyConsumer extends ProxyWrapper>> + implements Consumer> { + + @Autowired + public ProxyConsumer(ObjectMapper mapper) { + super(mapper); + } + + @Override + public void accept(Flux input) { + getDelegate().accept(input.map(this::fromJson)); } }