Twitter functional apps

* Streaming source
* Add twitter search and messasge sources
* Add more twitter source, sink and processors
* Add IT tests for twitter update sink
* Add IT tests for twitter message sink
* Add IT tests for twitter trend processor
* twitter suppliers readme
* twitter consumers readme
* twitter functions readme
* Disable TwitterStreamSourceTests

applications/processor/bridge-processor/README.adoc.tmp

@@ -0,0 +1,11 @@
+//tag::ref-doc[]
+= Bridge Processor
+
+A processor that bridges the input and ouput by simply passing the incoming payload to the outbound.
+
+=== Payload
+
+Any
+
+//end::ref-doc[]
+null

applications/processor/pom.xml

@@ -18,6 +18,7 @@
         <module>splitter-processor</module>
         <module>transform-processor</module>
         <module>script-processor</module>
+        <module>twitter-trend-processor</module>
     </modules>
 
 </project>

applications/processor/twitter-trend-processor/README.adoc

@@ -0,0 +1,36 @@
+//tag::ref-doc[]
+= Twitter Trend and Trend Locations Processor
+
+Processor that can return either trending topic or the Locations of the trending topics.
+The `twitter.trend.trend-query-type` property allow to select the query type.
+
+== Retrieve trending topic in a location (optionally)
+
+For this mode set `twitter.trend.trend-query-type` to `trend`.
+
+Processor based on https://developer.twitter.com/en/docs/trends/trends-for-location/api-reference/get-trends-place[Trends API].
+Returns the https://help.twitter.com/en/using-twitter/twitter-trending-faqs[trending topics] near a specific latitude, longitude location.
+
+== Retrieve trend Locations
+
+For this mode set `twitter.trend.trend-query-type` to `trendLocation`.
+
+Retrieve a full or nearby locations list of trending topics by location.
+
+If the `latitude`, `longitude` parameters are NOT provided the processor performs the https://developer.twitter.com/en/docs/trends/locations-with-trending-topics/api-reference/get-trends-available[Trends Available API] and returns the locations that Twitter has trending topic information for.
+
+If the `latitude`, `longitude` parameters are provided the processor performs the https://developer.twitter.com/en/docs/trends/locations-with-trending-topics/api-reference/get-trends-closest[Trends Closest API] and returns the locations that Twitter has trending topic information for, closest to a specified location.
+
+Response is an array of `locations` that encode the location's WOEID and some other human-readable information such as a canonical name and country the location belongs in.
+
+== Options
+
+//tag::configuration-properties[]
+$$twitter.trend.closest.lat$$:: $$If provided with a long parameter the available trend locations will be sorted by distance, nearest to furthest, to the co-ordinate pair. The valid ranges for longitude is -180.0 to +180.0 (West is negative, East is positive) inclusive.$$ *($$Expression$$, default: `$$<none>$$`)*
+$$twitter.trend.closest.lon$$:: $$If provided with a lat parameter the available trend locations will be sorted by distance, nearest to furthest, to the co-ordinate pair. The valid ranges for longitude is -180.0 to +180.0 (West is negative, East is positive) inclusive.$$ *($$Expression$$, default: `$$<none>$$`)*
+$$twitter.trend.location-id$$:: $$The Yahoo! Where On Earth ID of the location to return trending information for. Global information is available by using 1 as the WOEID.$$ *($$Expression$$, default: `$$payload$$`)*
+$$twitter.trend.trend-query-type$$:: $$<documentation missing>$$ *($$TrendQueryType$$, default: `$$<none>$$`, possible values: `trend`,`trendLocation`)*
+//end::configuration-properties[]
+
+//end::ref-doc[]
+

applications/processor/twitter-trend-processor/pom.xml

@@ -0,0 +1,103 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>twitter-trend-processor</artifactId>
+    <name>twitter-trend-processor</name>
+    <description>twitter trend processor apps</description>
+    <version>3.0.0-SNAPSHOT</version>
+
+    <parent>
+        <groupId>org.springframework.cloud.stream.app</groupId>
+        <artifactId>stream-applications-core</artifactId>
+        <version>3.0.0-SNAPSHOT</version>
+        <relativePath/>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>twitter-function</artifactId>
+            <version>${java-functions.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.junit.vintage</groupId>
+                    <artifactId>junit-vintage-engine</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-json</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mock-server</groupId>
+            <artifactId>mockserver-netty</artifactId>
+            <version>5.10</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mock-server</groupId>
+            <artifactId>mockserver-client-java</artifactId>
+            <version>5.10</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.cloud</groupId>
+                <artifactId>spring-cloud-app-starter-doc-maven-plugin</artifactId>
+            </plugin>
+            <plugin>
+                <groupId>org.springframework.cloud.stream.app.plugin</groupId>
+                <artifactId>spring-cloud-stream-app-maven-plugin</artifactId>
+                <configuration>
+                    <generatedApp>
+                        <name>twitter-trend</name>
+                        <type>processor</type>
+                        <version>${project.version}</version>
+                        <configClass>org.springframework.cloud.fn.twitter.trend.TwitterTrendFunctionConfiguration.class</configClass>
+                        <functionDefinition>trendOrTrendLocationsFunction</functionDefinition>
+                    </generatedApp>
+
+                    <dependencies>
+                        <dependency>
+                            <groupId>org.springframework.cloud.fn</groupId>
+                            <artifactId>twitter-function</artifactId>
+                            <version>${java-functions.version}</version>
+                        </dependency>
+                    </dependencies>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+    <repositories>
+        <repository>
+            <snapshots>
+                <enabled>true</enabled>
+            </snapshots>
+            <id>spring-snapshots</id>
+            <name>Spring Snapshots</name>
+            <url>https://repo.spring.io/libs-snapshot-local</url>
+        </repository>
+        <repository>
+            <snapshots>
+                <enabled>false</enabled>
+            </snapshots>
+            <id>spring-milestones</id>
+            <name>Spring Milestones</name>
+            <url>https://repo.spring.io/libs-milestone-local</url>
+        </repository>
+    </repositories>
+</project>

applications/processor/twitter-trend-processor/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties

@@ -0,0 +1,3 @@
+configuration-properties.classes=org.springframework.cloud.fn.twitter.trend.TwitterTrendFunctionProperties, \
+  org.springframework.cloud.fn.twitter.trend.TwitterTrendFunctionProperties$Closest
+

applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTestUtils.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2020-2020 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
+ *
+ *      https://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.processor.twitter.trend;
+
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.function.Function;
+
+import twitter4j.conf.ConfigurationBuilder;
+
+import org.springframework.core.io.DefaultResourceLoader;
+import org.springframework.util.StreamUtils;
+
+/**
+ * @author Christian Tzolov
+ */
+public class TwitterTestUtils {
+
+	public Function<ConfigurationBuilder, ConfigurationBuilder> mockTwitterUrls(String baseUrl) {
+		return configBuilder -> {
+			configBuilder.setRestBaseURL(baseUrl + "/");
+			configBuilder.setStreamBaseURL(baseUrl + "/stream/");
+			configBuilder.setUserStreamBaseURL(baseUrl + "/user/");
+			configBuilder.setSiteStreamBaseURL(baseUrl + "/site/");
+			configBuilder.setUploadBaseURL(baseUrl + "/upload/");
+
+			configBuilder.setOAuthAccessTokenURL(baseUrl + "/oauth/access_token");
+			configBuilder.setOAuthAuthenticationURL(baseUrl + "/oauth/authenticate");
+			configBuilder.setOAuthAuthorizationURL(baseUrl + "/oauth/authorize");
+			configBuilder.setOAuthRequestTokenURL(baseUrl + "/oauth/request_token");
+			configBuilder.setOAuth2TokenURL(baseUrl + "/oauth2/token");
+			configBuilder.setOAuth2InvalidateTokenURL(baseUrl + "/oauth2/invalidate_token");
+
+			return configBuilder;
+		};
+	}
+
+	/**
+	 * Load Spring Resource as String.
+	 * @param resourcePath Resource path (accepts file:// , classpath:// and http:// uri schemas)
+	 * @return Returns text (UTF8) representation of the resource pointed by the resourcePath
+	 */
+	public static String asString(String resourcePath) {
+		try {
+			return StreamUtils.copyToString(new DefaultResourceLoader().getResource(resourcePath).getInputStream(),
+					Charset.forName("UTF-8"));
+		}
+		catch (IOException e) {
+			throw new RuntimeException("Can not load resource:" + resourcePath, e);
+		}
+	}
+
+}

applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTrendLocationProcessorIntegrationTests.java

@@ -0,0 +1,226 @@
+/*
+ * Copyright 2020-2020 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
+ *
+ *      https://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.processor.twitter.trend;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockserver.client.MockServerClient;
+import org.mockserver.integration.ClientAndServer;
+import org.mockserver.model.Header;
+import org.mockserver.model.HttpRequest;
+import twitter4j.conf.ConfigurationBuilder;
+
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.fn.common.twitter.TwitterConnectionProperties;
+import org.springframework.cloud.fn.twitter.trend.TwitterTrendFunctionConfiguration;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Primary;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockserver.matchers.Times.exactly;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.HttpResponse.response;
+import static org.mockserver.verify.VerificationTimes.once;
+
+/**
+ * @author Christian Tzolov
+ */
+public class TwitterTrendLocationProcessorIntegrationTests {
+
+	private static final String MOCK_SERVER_IP = "127.0.0.1";
+
+	private static final Integer MOCK_SERVER_PORT = 1080;
+
+	private static ClientAndServer mockServer;
+
+	private static MockServerClient mockClient;
+	private static HttpRequest availableTrendsRequest;
+	private static HttpRequest closestTrendsRequest;
+
+	@BeforeEach
+	public void startServer() {
+		mockServer = ClientAndServer.startClientAndServer(MOCK_SERVER_PORT);
+		mockClient = new MockServerClient(MOCK_SERVER_IP, MOCK_SERVER_PORT);
+
+		availableTrendsRequest = setExpectation(request()
+				.withMethod("GET")
+				.withPath("/trends/available.json"));
+
+		closestTrendsRequest = setExpectation(request()
+				.withMethod("GET")
+				.withPath("/trends/closest.json")
+				.withQueryStringParameter("lat", "52.379189")
+				.withQueryStringParameter("long", "4.899431"));
+	}
+
+	@AfterEach
+	public void stopServer() {
+		mockServer.stop();
+	}
+
+	@Test
+	public void testTwitterAvailableTrends() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(TwitterTrendProcessorIntegrationTests.TestTwitterTrendProcessorApplication.class))
+				.web(WebApplicationType.NONE)
+				.run("--spring.cloud.stream.function.definition=trendOrTrendLocationsFunction",
+
+						"--twitter.trend.trendQueryType=trendLocation",
+						"--twitter.connection.rawJson=false",
+
+						"--twitter.connection.consumerKey=myConsumerKey",
+						"--twitter.connection.consumerSecret=myConsumerSecret",
+						"--twitter.connection.accessToken=myAccessToken",
+						"--twitter.connection.accessTokenSecret=myAccessTokenSecret")) {
+
+			InputDestination input = context.getBean(InputDestination.class);
+			OutputDestination output = context.getBean(OutputDestination.class);
+			assertThat(input).isNotNull();
+			assertThat(output).isNotNull();
+
+			input.send(new GenericMessage<>("hello".getBytes(StandardCharsets.UTF_8)));
+
+			Message<byte[]> outputMessage = output.receive(Duration.ofSeconds(300).toMillis());
+			assertThat(outputMessage).isNotNull();
+
+			mockClient.verify(availableTrendsRequest, once());
+			assertThat(outputMessage);
+
+			String payload = new String(outputMessage.getPayload());
+			assertThat(payload).containsSequence("countryName");
+			assertThat(payload).contains("placeCode");
+			assertThat(payload).doesNotContain("placeType");
+		}
+	}
+
+	@Test
+	public void testTwitterAvailableTrendsTwitterJson() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(TwitterTrendProcessorIntegrationTests.TestTwitterTrendProcessorApplication.class))
+				.web(WebApplicationType.NONE)
+				.run("--spring.cloud.stream.function.definition=trendOrTrendLocationsFunction",
+
+						"--twitter.trend.trendQueryType=trendLocation",
+						"--twitter.connection.rawJson=true",
+
+						"--twitter.connection.consumerKey=myConsumerKey",
+						"--twitter.connection.consumerSecret=myConsumerSecret",
+						"--twitter.connection.accessToken=myAccessToken",
+						"--twitter.connection.accessTokenSecret=myAccessTokenSecret")) {
+
+			InputDestination input = context.getBean(InputDestination.class);
+			OutputDestination output = context.getBean(OutputDestination.class);
+			assertThat(input).isNotNull();
+			assertThat(output).isNotNull();
+
+			input.send(new GenericMessage<>("hello".getBytes(StandardCharsets.UTF_8)));
+
+			Message<byte[]> outputMessage = output.receive(Duration.ofSeconds(300).toMillis());
+			assertThat(outputMessage).isNotNull();
+
+			mockClient.verify(availableTrendsRequest, once());
+			assertThat(outputMessage).isNotNull();
+
+			String payload = new String(outputMessage.getPayload());
+
+			assertThat(payload).contains("placeType");
+			assertThat(payload).doesNotContain("placeCode");
+			assertThat(payload).doesNotContain("countryName");
+		}
+	}
+
+	@Test
+	public void testTwitterClosestTrends() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(TwitterTrendProcessorIntegrationTests.TestTwitterTrendProcessorApplication.class))
+				.web(WebApplicationType.NONE)
+				.run("--spring.cloud.stream.function.definition=trendOrTrendLocationsFunction",
+
+						"--twitter.trend.trendQueryType=trendLocation",
+						"--twitter.connection.rawJson=true",
+						"--twitter.trend.closest.lat='52.379189'",
+						"--twitter.trend.closest.lon='4.899431'",
+
+						"--twitter.connection.consumerKey=myConsumerKey",
+						"--twitter.connection.consumerSecret=myConsumerSecret",
+						"--twitter.connection.accessToken=myAccessToken",
+						"--twitter.connection.accessTokenSecret=myAccessTokenSecret")) {
+
+			InputDestination input = context.getBean(InputDestination.class);
+			OutputDestination output = context.getBean(OutputDestination.class);
+			assertThat(input).isNotNull();
+			assertThat(output).isNotNull();
+
+			input.send(new GenericMessage<>("hello".getBytes(StandardCharsets.UTF_8)));
+
+			Message<byte[]> outputMessage = output.receive(Duration.ofSeconds(300).toMillis());
+			assertThat(outputMessage).isNotNull();
+
+			mockClient.verify(closestTrendsRequest, once());
+			assertThat(outputMessage).isNotNull();
+		}
+	}
+
+	public static HttpRequest setExpectation(HttpRequest request) {
+		mockClient
+				.when(request, exactly(1))
+				.respond(response()
+						.withStatusCode(200)
+						.withHeaders(
+								new Header("Content-Type", "application/json; charset=utf-8"),
+								new Header("Cache-Control", "public, max-age=86400"))
+						.withBody(TwitterTestUtils.asString("classpath:/response/trend_locations.json"))
+						.withDelay(TimeUnit.SECONDS, 1)
+				);
+		return request;
+	}
+
+	@SpringBootConfiguration
+	@EnableAutoConfiguration
+	@Import(TwitterTrendFunctionConfiguration.class)
+	public static class TestTwitterTrendLocationProcessorApplication {
+		@Bean
+		@Primary
+		public twitter4j.conf.Configuration twitterConfiguration2(TwitterConnectionProperties properties,
+				Function<TwitterConnectionProperties, ConfigurationBuilder> toConfigurationBuilder) {
+
+			Function<TwitterConnectionProperties, ConfigurationBuilder> mockedConfiguration =
+					toConfigurationBuilder.andThen(
+							new TwitterTestUtils().mockTwitterUrls(
+									String.format("http://%s:%s", MOCK_SERVER_IP, MOCK_SERVER_PORT)));
+
+			return mockedConfiguration.apply(properties).build();
+		}
+	}
+
+}

applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTrendProcessorIntegrationTests.java

@@ -0,0 +1,151 @@
+/*
+ * Copyright 2020-2020 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
+ *
+ *      https://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.processor.twitter.trend;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockserver.client.MockServerClient;
+import org.mockserver.integration.ClientAndServer;
+import org.mockserver.model.Header;
+import org.mockserver.model.HttpRequest;
+import twitter4j.conf.ConfigurationBuilder;
+
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.fn.common.twitter.TwitterConnectionProperties;
+import org.springframework.cloud.fn.twitter.trend.TwitterTrendFunctionConfiguration;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Primary;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockserver.matchers.Times.exactly;
+import static org.mockserver.model.HttpRequest.request;
+import static org.mockserver.model.HttpResponse.response;
+import static org.mockserver.verify.VerificationTimes.once;
+
+/**
+ * @author Christian Tzolov
+ */
+public class TwitterTrendProcessorIntegrationTests {
+
+	private static final String MOCK_SERVER_IP = "127.0.0.1";
+
+	private static final Integer MOCK_SERVER_PORT = 1080;
+
+	private static ClientAndServer mockServer;
+
+	private static MockServerClient mockClient;
+	private static HttpRequest trendsRequest;
+
+	@BeforeAll
+	public static void startServer() {
+		mockServer = ClientAndServer.startClientAndServer(MOCK_SERVER_PORT);
+		mockClient = new MockServerClient(MOCK_SERVER_IP, MOCK_SERVER_PORT);
+
+		trendsRequest = setExpectation(request()
+				.withMethod("GET")
+				.withPath("/trends/place.json")
+				.withQueryStringParameter("id", "2972"));
+	}
+
+	@AfterAll
+	public static void stopServer() {
+		mockServer.stop();
+	}
+
+	@Test
+	public void testTwitterTrendPayload() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+				TestChannelBinderConfiguration.getCompleteConfiguration(TestTwitterTrendProcessorApplication.class))
+				.web(WebApplicationType.NONE)
+				.run("--spring.cloud.stream.function.definition=trendOrTrendLocationsFunction",
+
+						"--twitter.trend.locationId='2972'",
+						"--twitter.connection.rawJson=true",
+
+						"--twitter.connection.consumerKey=myConsumerKey",
+						"--twitter.connection.consumerSecret=myConsumerSecret",
+						"--twitter.connection.accessToken=myAccessToken",
+						"--twitter.connection.accessTokenSecret=myAccessTokenSecret")) {
+
+			InputDestination input = context.getBean(InputDestination.class);
+			OutputDestination output = context.getBean(OutputDestination.class);
+			assertThat(input).isNotNull();
+			assertThat(output).isNotNull();
+
+			input.send(new GenericMessage<>("Hello".getBytes(StandardCharsets.UTF_8)));
+
+			Message<byte[]> outputMessage = output.receive(Duration.ofSeconds(300).toMillis());
+			assertThat(outputMessage).isNotNull();
+
+			mockClient.verify(trendsRequest, once());
+
+			//Resource trendsResource = new DefaultResourceLoader().getResource("classpath:/response/trends.json");
+			//String expected = new String(StreamUtils.copyToByteArray(trendsResource.getInputStream()), StandardCharsets.UTF_8).trim();
+			//String actual = new String(outputMessage.getPayload(), StandardCharsets.UTF_8);
+			//JSONAssert.assertEquals(expected, actual, JSONCompareMode.LENIENT);
+		}
+	}
+
+	public static HttpRequest setExpectation(HttpRequest request) {
+		mockClient
+				.when(request, exactly(1))
+				.respond(response()
+						.withStatusCode(200)
+						.withHeaders(
+								new Header("Content-Type", "application/json; charset=utf-8"),
+								new Header("Cache-Control", "public, max-age=86400"))
+						.withBody(TwitterTestUtils.asString("classpath:/response/trends.json"))
+						.withDelay(TimeUnit.SECONDS, 1)
+				);
+		return request;
+	}
+
+	@SpringBootConfiguration
+	@EnableAutoConfiguration
+	@Import(TwitterTrendFunctionConfiguration.class)
+	public static class TestTwitterTrendProcessorApplication {
+		@Bean
+		@Primary
+		public twitter4j.conf.Configuration twitterConfiguration2(TwitterConnectionProperties properties,
+				Function<TwitterConnectionProperties, ConfigurationBuilder> toConfigurationBuilder) {
+
+			Function<TwitterConnectionProperties, ConfigurationBuilder> mockedConfiguration =
+					toConfigurationBuilder.andThen(
+							new TwitterTestUtils().mockTwitterUrls(
+									String.format("http://%s:%s", MOCK_SERVER_IP, MOCK_SERVER_PORT)));
+
+			return mockedConfiguration.apply(properties).build();
+		}
+	}
+
+}