From 9983453b777a6e091898c415c19e22e47c6f7e77 Mon Sep 17 00:00:00 2001
From: Janne Valkealahti <janne.valkealahti@gmail.com>
Date: Thu, 12 Jan 2017 08:37:10 +0000
Subject: [PATCH] Add target caching to BindableProxyFactory

- Prevent new reflection calls to channels
  for Sources, Sinks and Processors, cache bound target
  if it's found and try to use it next time as target
  should not change.
- One test for checking caching.
- Fixes #759
---
 .../stream/binding/BindableProxyFactory.java  | 19 ++++++--
 .../stream/aggregation/AggregationTest.java   | 43 ++++++++++++++++++-
 2 files changed, 58 insertions(+), 4 deletions(-)

diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java
index 0ba5a7315..acf6167db 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -74,6 +74,8 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
 
 	private Map<String, BoundTargetHolder> outputHolders = new HashMap<>();
 
+	private final Map<Method, Object> targetCache = new HashMap<>(2);
+
 	public BindableProxyFactory(Class<?> type) {
 		this.type = type;
 	}
@@ -81,16 +83,27 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
 	@Override
 	public synchronized Object invoke(MethodInvocation invocation) throws Throwable {
 		Method method = invocation.getMethod();
+
+		// try to use cached target
+		Object boundTarget = targetCache.get(method);
+		if (boundTarget != null) {
+			return boundTarget;
+		}
+
 		Input input = AnnotationUtils.findAnnotation(method, Input.class);
 		if (input != null) {
 			String name = BindingBeanDefinitionRegistryUtils.getBindingTargetName(input, method);
-			return this.inputHolders.get(name).getBoundTarget();
+			boundTarget =  this.inputHolders.get(name).getBoundTarget();
+			targetCache.put(method, boundTarget);
+			return boundTarget;
 		}
 		else {
 			Output output = AnnotationUtils.findAnnotation(method, Output.class);
 			if (output != null) {
 				String name = BindingBeanDefinitionRegistryUtils.getBindingTargetName(output, method);
-				return this.outputHolders.get(name).getBoundTarget();
+				boundTarget = this.outputHolders.get(name).getBoundTarget();
+				targetCache.put(method, boundTarget);
+				return boundTarget;
 			}
 		}
 		return null;
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java
index 92053d122..f310ea322 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,11 +16,14 @@
 
 package org.springframework.cloud.stream.aggregation;
 
+import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Test;
 
 import org.springframework.beans.DirectFieldAccessor;
@@ -30,13 +33,16 @@ import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder.So
 import org.springframework.cloud.stream.aggregate.SharedBindingTargetRegistry;
 import org.springframework.cloud.stream.aggregate.SharedChannelRegistry;
 import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.binding.BindableProxyFactory;
 import org.springframework.cloud.stream.binding.BindingTargetFactory;
 import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.util.ReflectionUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertTrue;
@@ -331,6 +337,41 @@ public class AggregationTest {
 		aggregatedApplicationContext.close();
 	}
 
+	@Test
+	public void testBindableProxyFactoryCaching() {
+		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(MockBinderRegistryConfiguration.class,
+				TestSource.class, TestProcessor.class);
+		Map<String, BindableProxyFactory> factories = context.getBeansOfType(BindableProxyFactory.class);
+		assertThat(factories).hasSize(2);
+
+		Map<String, Source> sources = context.getBeansOfType(Source.class);
+		assertThat(sources).hasSize(2);
+		for (Source source : sources.values()) {
+			source.output();
+		}
+
+		Map<String, Processor> processors = context.getBeansOfType(Processor.class);
+		assertThat(processors).hasSize(1);
+		for (Processor processor : processors.values()) {
+			processor.input();
+			processor.output();
+		}
+
+		for (BindableProxyFactory factory : factories.values()) {
+			Field field = ReflectionUtils.findField(BindableProxyFactory.class, "targetCache");
+			ReflectionUtils.makeAccessible(field);
+			Map<?, ?> targetCache = (Map<?, ?>) ReflectionUtils.getField(field, factory);
+			if (factory.getObjectType() == Source.class) {
+				assertThat(targetCache).hasSize(1);
+			} else if (factory.getObjectType() == Processor.class) {
+				assertThat(targetCache).hasSize(2);
+			} else {
+				Assert.fail("Found unexpected type");
+			}
+		}
+		context.close();
+	}
+
 	@EnableBinding(Source.class)
 	@EnableAutoConfiguration
 	public static class TestSource {