/*
 * Copyright 2015 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.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * 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 org.springframework.cloud.stream.app.gemfire.source;

import com.gemstone.gemfire.cache.Region;
import com.gemstone.gemfire.pdx.PdxInstance;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.app.gemfire.JsonObjectTransformer;
import org.springframework.cloud.stream.app.gemfire.config.GemfireClientRegionConfiguration;
import org.springframework.cloud.stream.app.gemfire.config.GemfirePoolConfiguration;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.gemfire.inbound.CacheListeningMessageProducer;
import org.springframework.integration.router.PayloadTypeRouter;
import org.springframework.integration.transformer.GenericTransformer;
import org.springframework.messaging.MessageChannel;

import javax.annotation.Resource;

/**
 * The Gemfire Source provides a {@link CacheListeningMessageProducer} which produces a
 * message for cache events. Internally it uses a {@link com.gemstone.gemfire.cache.CacheListener}
 * which, by default, emits a native GemFire type which holds all of the event details.
 * This is not ideal for streaming applications because it require this type to
 * be also in the consuming app's classpath. Hence, a SpEl Expression, given by the
 * property 'cacheEventExpression' is used to extract required information from the
 * payload. The default expression is 'newValue' which returns the updated object. This
 * may not be ideal for every use case especially since it does not provide the key
 * value. The key is referenced by the field 'key'. Also available are 'operation' (the
 * operation associated with the event, and 'oldValue'. If the cached key and value
 * types are primitives, an simple expression like "'key' + key + ':' + value" may work.
 *
 * More complex transformations, such as Json, will require customization. To access
 * the original object, set 'cacheEntryExpression' to '#root' or "#this'.
 *
 * This converts payloads of type {@link PdxInstance}, which Gemfire uses to store
 * JSON content, to a JSON String.
 *
 *
 * @author David Turanski
 */
@EnableBinding(Source.class)
@Import({ KeyInterestConfiguration.class,
		GemfirePoolConfiguration.class,
		GemfireClientRegionConfiguration.class })
@EnableConfigurationProperties(GemfireSourceProperties.class)
@PropertySource("classpath:gemfire-source.yml")
public class GemfireSource {

	@Autowired
	private GemfireSourceProperties config;

	//NOTE: https://jira.spring.io/browse/SPR-7915 supposedly fixed in SF 4.3. So
	//should be able to change to @Autowired at that point
	@Resource(name = "clientRegion")
	private Region<String, ?> region;

	@Autowired
	@Qualifier(Source.OUTPUT)
	private MessageChannel output;

	@Bean
	public MessageChannel convertToStringChannel(){
		return new DirectChannel();
	}

	@Bean
	public MessageChannel toRouterChannel(){
		return new DirectChannel();
	}

	@Bean
	PayloadTypeRouter payloadTypeRouter(){
		PayloadTypeRouter router = new PayloadTypeRouter();
		router.setDefaultOutputChannel(output);
		router.setChannelMapping(PdxInstance.class.getName(),"convertToStringChannel");

		return router;
	}

	@Bean
	public IntegrationFlow startFlow() {
		return IntegrationFlows.from(toRouterChannel()).route
				(payloadTypeRouter())
				.get();
	}

	@Bean JsonObjectTransformer transformer() {
		return new JsonObjectTransformer();
	}

	@Bean IntegrationFlow convertToString() {
		return IntegrationFlows.from(convertToStringChannel())
				.transform(new GenericTransformer<PdxInstance, String>() {
					@Override public String transform(PdxInstance pdxInstance) {
						return transformer().toString(pdxInstance);
					}
				})
				.channel(output)
				.get();
	}

	@Bean
	public CacheListeningMessageProducer cacheListeningMessageProducer() {
		CacheListeningMessageProducer cacheListeningMessageProducer = new
				CacheListeningMessageProducer(region);
		cacheListeningMessageProducer.setOutputChannel(toRouterChannel());
		cacheListeningMessageProducer.setExpressionPayload(
				config.getCacheEventExpression());
		return cacheListeningMessageProducer;
	}

}

@@ -1847,79 +1847,8 @@
 er;%0A
-import org.springframework.integration.transformer.GenericTransformer;%0A
 impo
@@ -3951,19 +3951,17 @@
 Channel 
-toR
+r
 outerCha
@@ -4325,19 +4325,17 @@
 ws.from(
-toR
+r
 outerCha
@@ -4345,19 +4345,19 @@
 l())
+%0A%09%09%09%09
 .route
-%0A%09%09%09%09
 (pay
@@ -4596,135 +4596,8 @@
 orm(
-new GenericTransformer%3CPdxInstance, String%3E() %7B%0A%09%09%09%09%09@Override public String transform(PdxInstance pdxInstance) %7B%0A%09%09%09%09%09%09return 
 tran
@@ -4609,17 +4609,18 @@
 er()
-.
+,%22
 toString
 (pdx
@@ -4619,35 +4619,9 @@
 ring
-(pdxInstance);%0A%09%09%09%09%09%7D%0A%09%09%09%09%7D
+%22
 )%0A%09%09
@@ -4898,11 +4898,9 @@
 nel(
-toR
+r
 oute

Remove GenericTransformer