diff --git a/applications/processor/twitter-trend-processor/pom.xml b/applications/processor/twitter-trend-processor/pom.xml index 1340356b..40ac51db 100644 --- a/applications/processor/twitter-trend-processor/pom.xml +++ b/applications/processor/twitter-trend-processor/pom.xml @@ -67,7 +67,7 @@ processor ${project.version} org.springframework.cloud.fn.twitter.trend.TwitterTrendFunctionConfiguration.class - trendOrTrendLocationsFunction + twitterTrendFunction diff --git a/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTestUtils.java b/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTestUtils.java deleted file mode 100644 index b76e8e30..00000000 --- a/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTestUtils.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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 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); - } - } - -} diff --git a/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTrendLocationProcessorIntegrationTests.java b/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTrendLocationProcessorIntegrationTests.java index afafefb6..5741a4b7 100644 --- a/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTrendLocationProcessorIntegrationTests.java +++ b/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTrendLocationProcessorIntegrationTests.java @@ -35,6 +35,7 @@ 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.common.twitter.util.TwitterTestUtils; import org.springframework.cloud.fn.twitter.trend.TwitterTrendFunctionConfiguration; import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.OutputDestination; @@ -45,6 +46,7 @@ import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Primary; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.SocketUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.mockserver.matchers.Times.exactly; @@ -59,7 +61,7 @@ public class TwitterTrendLocationProcessorIntegrationTests { private static final String MOCK_SERVER_IP = "127.0.0.1"; - private static final Integer MOCK_SERVER_PORT = 1080; + private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort(); private static ClientAndServer mockServer; @@ -91,9 +93,9 @@ public class TwitterTrendLocationProcessorIntegrationTests { @Test public void testTwitterAvailableTrends() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(TwitterTrendProcessorIntegrationTests.TestTwitterTrendProcessorApplication.class)) + TestChannelBinderConfiguration.getCompleteConfiguration(TestTwitterTrendLocationProcessorApplication.class)) .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=trendOrTrendLocationsFunction", + .run("--spring.cloud.stream.function.definition=twitterTrendFunction", "--twitter.trend.trendQueryType=trendLocation", "--twitter.connection.rawJson=false", @@ -126,9 +128,9 @@ public class TwitterTrendLocationProcessorIntegrationTests { @Test public void testTwitterAvailableTrendsTwitterJson() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(TwitterTrendProcessorIntegrationTests.TestTwitterTrendProcessorApplication.class)) + TestChannelBinderConfiguration.getCompleteConfiguration(TestTwitterTrendLocationProcessorApplication.class)) .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=trendOrTrendLocationsFunction", + .run("--spring.cloud.stream.function.definition=twitterTrendFunction", "--twitter.trend.trendQueryType=trendLocation", "--twitter.connection.rawJson=true", @@ -162,9 +164,9 @@ public class TwitterTrendLocationProcessorIntegrationTests { @Test public void testTwitterClosestTrends() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(TwitterTrendProcessorIntegrationTests.TestTwitterTrendProcessorApplication.class)) + TestChannelBinderConfiguration.getCompleteConfiguration(TestTwitterTrendLocationProcessorApplication.class)) .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=trendOrTrendLocationsFunction", + .run("--spring.cloud.stream.function.definition=twitterTrendFunction", "--twitter.trend.trendQueryType=trendLocation", "--twitter.connection.rawJson=true", diff --git a/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTrendProcessorIntegrationTests.java b/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTrendProcessorIntegrationTests.java index d6c6c9cb..515e6443 100644 --- a/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTrendProcessorIntegrationTests.java +++ b/applications/processor/twitter-trend-processor/src/test/java/org/springframework/cloud/stream/app/processor/twitter/trend/TwitterTrendProcessorIntegrationTests.java @@ -35,6 +35,7 @@ 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.common.twitter.util.TwitterTestUtils; import org.springframework.cloud.fn.twitter.trend.TwitterTrendFunctionConfiguration; import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.OutputDestination; @@ -45,6 +46,7 @@ import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Primary; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.SocketUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.mockserver.matchers.Times.exactly; @@ -59,7 +61,7 @@ public class TwitterTrendProcessorIntegrationTests { private static final String MOCK_SERVER_IP = "127.0.0.1"; - private static final Integer MOCK_SERVER_PORT = 1080; + private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort(); private static ClientAndServer mockServer; @@ -87,7 +89,7 @@ public class TwitterTrendProcessorIntegrationTests { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration(TestTwitterTrendProcessorApplication.class)) .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=trendOrTrendLocationsFunction", + .run("--spring.cloud.stream.function.definition=twitterTrendFunction", "--twitter.trend.locationId='2972'", "--twitter.connection.rawJson=true", diff --git a/applications/sink/twitter-message-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/message/TwitterMessageSinkIntegrationTests.java b/applications/sink/twitter-message-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/message/TwitterMessageSinkIntegrationTests.java index 09117888..4fc332d4 100644 --- a/applications/sink/twitter-message-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/message/TwitterMessageSinkIntegrationTests.java +++ b/applications/sink/twitter-message-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/message/TwitterMessageSinkIntegrationTests.java @@ -35,6 +35,7 @@ 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.common.twitter.util.TwitterTestUtils; import org.springframework.cloud.fn.consumer.twitter.message.TwitterMessageConsumerConfiguration; import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; @@ -43,6 +44,7 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Primary; import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.SocketUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.mockserver.matchers.Times.unlimited; @@ -57,7 +59,7 @@ public class TwitterMessageSinkIntegrationTests { private static final String MOCK_SERVER_IP = "127.0.0.1"; - private static final Integer MOCK_SERVER_PORT = 1080; + private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort(); private static ClientAndServer mockServer; diff --git a/applications/sink/twitter-message-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/message/TwitterTestUtils.java b/applications/sink/twitter-message-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/message/TwitterTestUtils.java deleted file mode 100644 index 2507022e..00000000 --- a/applications/sink/twitter-message-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/message/TwitterTestUtils.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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.sink.twitter.message; - - -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 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); - } - } - -} diff --git a/applications/sink/twitter-update-sink/README.adoc b/applications/sink/twitter-update-sink/README.adoc index 12b0f8cd..d4e0e6d6 100644 --- a/applications/sink/twitter-update-sink/README.adoc +++ b/applications/sink/twitter-update-sink/README.adoc @@ -32,13 +32,13 @@ $$twitter.update.text$$:: $$(SpEL expression) The text of the text update. URL e `TwitterUpdateConsumerConfiguration` exposes 2 composable functions: -* `Function, StatusUpdate> toStatusUpdateQuery(TwitterUpdateConsumerProperties updateProperties)` - Converts input message into `StatusUpdate` query object. +* `Function, StatusUpdate> messageToStatusUpdateFunction(TwitterUpdateConsumerProperties updateProperties)` - Converts input message into `StatusUpdate` query object. * `Consumer updateStatus(Twitter twitter)` - Sends the input `StatusUpdate` argument as Twitter text update. Use `@Import(TwitterUpdateConsumerConfiguration.class)` to compose those functions. By default the `twitter-update` implements the following composite function chain: -`spring.cloud.stream.function.definition=byteArrayTextToString|toStatusUpdateQuery|updateStatus` or (`byteArrayTextToString|twitterStatusUpdateConsumer`) +`spring.cloud.stream.function.definition=byteArrayTextToString|messageToStatusUpdateFunction|updateStatus` or (`byteArrayTextToString|twitterStatusUpdateConsumer`) == Examples diff --git a/applications/sink/twitter-update-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/update/TwitterUpdateSinkIntegrationTests.java b/applications/sink/twitter-update-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/update/TwitterUpdateSinkIntegrationTests.java index ca743f26..0d7a9306 100644 --- a/applications/sink/twitter-update-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/update/TwitterUpdateSinkIntegrationTests.java +++ b/applications/sink/twitter-update-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/update/TwitterUpdateSinkIntegrationTests.java @@ -33,6 +33,7 @@ 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.common.twitter.util.TwitterTestUtils; import org.springframework.cloud.fn.consumer.twitter.status.update.TwitterUpdateConsumerConfiguration; import org.springframework.cloud.fn.consumer.twitter.status.update.TwitterUpdateConsumerProperties; import org.springframework.cloud.stream.binder.test.InputDestination; @@ -42,6 +43,7 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Primary; import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.SocketUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.mockserver.matchers.Times.unlimited; @@ -56,7 +58,7 @@ public class TwitterUpdateSinkIntegrationTests { private static final String MOCK_SERVER_IP = "127.0.0.1"; - private static final Integer MOCK_SERVER_PORT = 1080; + private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort(); private static ClientAndServer mockServer; diff --git a/applications/source/twitter-message-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/message/TwitterMessageSourceIntegrationTests.java b/applications/source/twitter-message-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/message/TwitterMessageSourceIntegrationTests.java index 47d2cccd..f58a79b2 100644 --- a/applications/source/twitter-message-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/message/TwitterMessageSourceIntegrationTests.java +++ b/applications/source/twitter-message-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/message/TwitterMessageSourceIntegrationTests.java @@ -37,6 +37,7 @@ 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.common.twitter.util.TwitterTestUtils; import org.springframework.cloud.fn.supplier.twitter.message.TwitterMessageSupplierConfiguration; import org.springframework.cloud.fn.supplier.twitter.message.TwitterMessageSupplierProperties; import org.springframework.cloud.stream.binder.test.OutputDestination; @@ -47,6 +48,7 @@ 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.util.SocketUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.mockserver.matchers.Times.exactly; @@ -61,7 +63,7 @@ public class TwitterMessageSourceIntegrationTests { private static final String MOCK_SERVER_IP = "127.0.0.1"; - private static final Integer MOCK_SERVER_PORT = 1080; + private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort(); private static ClientAndServer mockServer; diff --git a/applications/source/twitter-message-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/message/TwitterTestUtils.java b/applications/source/twitter-message-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/message/TwitterTestUtils.java deleted file mode 100644 index a314ef3a..00000000 --- a/applications/source/twitter-message-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/message/TwitterTestUtils.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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.source.twitter.message; - - -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 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); - } - } - -} diff --git a/applications/source/twitter-search-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/search/TwitterSearchSourceIntegrationTests.java b/applications/source/twitter-search-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/search/TwitterSearchSourceIntegrationTests.java index 5cea47d3..e7b4fca3 100644 --- a/applications/source/twitter-search-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/search/TwitterSearchSourceIntegrationTests.java +++ b/applications/source/twitter-search-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/search/TwitterSearchSourceIntegrationTests.java @@ -37,6 +37,7 @@ 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.common.twitter.util.TwitterTestUtils; import org.springframework.cloud.fn.supplier.twitter.status.search.TwitterSearchSupplierConfiguration; import org.springframework.cloud.fn.supplier.twitter.status.search.TwitterSearchSupplierProperties; import org.springframework.cloud.stream.binder.test.OutputDestination; @@ -47,6 +48,7 @@ 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.util.SocketUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.mockserver.matchers.Times.exactly; @@ -61,7 +63,7 @@ public class TwitterSearchSourceIntegrationTests { private static final String MOCK_SERVER_IP = "127.0.0.1"; - private static final Integer MOCK_SERVER_PORT = 1080; + private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort(); private static ClientAndServer mockServer; diff --git a/applications/source/twitter-search-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/search/TwitterTestUtils.java b/applications/source/twitter-search-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/search/TwitterTestUtils.java deleted file mode 100644 index 3f6c36ba..00000000 --- a/applications/source/twitter-search-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/search/TwitterTestUtils.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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.source.twitter.search; - - -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 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); - } - } - -} diff --git a/applications/source/twitter-stream-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/stream/TwitterStreamSourceTests.java b/applications/source/twitter-stream-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/stream/TwitterStreamSourceTests.java index 33186aba..d5e5b1a3 100644 --- a/applications/source/twitter-stream-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/stream/TwitterStreamSourceTests.java +++ b/applications/source/twitter-stream-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/stream/TwitterStreamSourceTests.java @@ -35,6 +35,7 @@ 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.common.twitter.util.TwitterTestUtils; import org.springframework.cloud.fn.supplier.twitter.status.stream.TwitterStreamSupplierConfiguration; import org.springframework.cloud.fn.supplier.twitter.status.stream.TwitterStreamSupplierProperties; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; @@ -42,6 +43,7 @@ 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.util.SocketUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.mockserver.matchers.Times.exactly; @@ -53,7 +55,7 @@ public class TwitterStreamSourceTests { private static final String MOCK_SERVER_IP = "127.0.0.1"; - private static final Integer MOCK_SERVER_PORT = 1080; + private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort(); private static ClientAndServer mockServer; diff --git a/applications/source/twitter-stream-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/stream/TwitterTestUtils.java b/applications/source/twitter-stream-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/stream/TwitterTestUtils.java deleted file mode 100644 index dbf96b3c..00000000 --- a/applications/source/twitter-stream-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/stream/TwitterTestUtils.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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.source.twitter.stream; - - -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 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); - } - } - -} diff --git a/functions/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionConfiguration.java b/functions/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionConfiguration.java index 057204d2..b5b65c78 100644 --- a/functions/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionConfiguration.java +++ b/functions/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionConfiguration.java @@ -39,9 +39,6 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.MimeTypeUtils; -//import org.springframework.cloud.stream.config.BindingProperties; -// import org.springframework.integration.support.MutableMessage; - /** * * @author Christian Tzolov diff --git a/applications/sink/twitter-update-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/update/TwitterTestUtils.java b/functions/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/util/TwitterTestUtils.java similarity index 97% rename from applications/sink/twitter-update-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/update/TwitterTestUtils.java rename to functions/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/util/TwitterTestUtils.java index 0ed73f54..25b27679 100644 --- a/applications/sink/twitter-update-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/update/TwitterTestUtils.java +++ b/functions/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/util/TwitterTestUtils.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.app.sink.twitter.update; +package org.springframework.cloud.fn.common.twitter.util; import java.io.IOException; diff --git a/functions/consumer/twitter-consumer/README.adoc b/functions/consumer/twitter-consumer/README.adoc index 2e95ceac..1f95e184 100644 --- a/functions/consumer/twitter-consumer/README.adoc +++ b/functions/consumer/twitter-consumer/README.adoc @@ -22,9 +22,9 @@ You can import `TwitterUpdateConsumerConfiguration` in the application and then - `Consumer updateStatus` - if you have an `StatusUpdate` instance you can use the `updateStatus` to apply it. -- `Function, StatusUpdate> toStatusUpdateQuery` - function that converts a `Message` text into a `StatusUpdate` instance using the `TwitterUpdateConsumerProperties` properties. +- `Function, StatusUpdate> messageToStatusUpdateFunction` - function that converts a `Message` text into a `StatusUpdate` instance using the `TwitterUpdateConsumerProperties` properties. -- `Consumer> twitterStatusUpdateConsumer` - composes `toStatusUpdateQuery` and `updateStatus` to update the twitter status from Message text. +- `Consumer> twitterStatusUpdateConsumer` - composes `messageToStatusUpdateFunction` and `updateStatus` to update the twitter status from Message text. Note: the Message content is expected to be in text format. Consider using the `byteArrayTextToString` utility `Function`. diff --git a/functions/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerConfiguration.java b/functions/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerConfiguration.java index cb301826..af49801d 100644 --- a/functions/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerConfiguration.java +++ b/functions/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerConfiguration.java @@ -16,7 +16,6 @@ package org.springframework.cloud.fn.consumer.twitter.status.update; -import java.util.Properties; import java.util.function.Consumer; import java.util.function.Function; @@ -28,16 +27,12 @@ import twitter4j.StatusUpdate; import twitter4j.Twitter; import twitter4j.TwitterException; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.fn.common.twitter.TwitterConnectionConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; -import org.springframework.core.env.ConfigurableEnvironment; -import org.springframework.core.env.PropertiesPropertySource; import org.springframework.messaging.Message; -import org.springframework.util.Assert; /** * @@ -50,13 +45,6 @@ public class TwitterUpdateConsumerConfiguration { private static final Log logger = LogFactory.getLog(TwitterUpdateConsumerConfiguration.class); - @Autowired - public void setInfoProperties(ConfigurableEnvironment env) { - Properties props = new Properties(); - props.put("spring.cloud.stream.function.definition", "toText|upper|sink"); - env.getPropertySources().addFirst(new PropertiesPropertySource("function-dsl-props", props)); - } - @Bean public Consumer updateStatus(Twitter twitter) { return statusUpdate -> { @@ -74,7 +62,8 @@ public class TwitterUpdateConsumerConfiguration { } @Bean - public Function, StatusUpdate> toStatusUpdateQuery(TwitterUpdateConsumerProperties updateProperties) { + public Function, StatusUpdate> messageToStatusUpdateFunction( + TwitterUpdateConsumerProperties updateProperties) { return message -> { @@ -106,8 +95,6 @@ public class TwitterUpdateConsumerConfiguration { } if (updateProperties.getLocation().getLat() != null) { - Assert.notNull(updateProperties.getLocation().getLon(), - "If the latitude is set then the longitude must be set too"); double lat = updateProperties.getLocation().getLat().getValue(message, Double.class); double lon = updateProperties.getLocation().getLon().getValue(message, Double.class); statusUpdate.setLocation(new GeoLocation(lat, lon)); diff --git a/functions/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerProperties.java b/functions/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerProperties.java index 5ce35d7f..a1b30e6c 100644 --- a/functions/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerProperties.java +++ b/functions/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerProperties.java @@ -81,7 +81,7 @@ public class TwitterUpdateConsumerProperties { /** * (SpEL expression) The location this Tweet refers to. Ignored if geo_enabled for the user is false! */ - private Location location = new Location(); + private final Location location = new Location(); public Expression getText() { return text; diff --git a/functions/consumer/twitter-consumer/src/test/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateSinkFunctionConfigurationTests.java b/functions/consumer/twitter-consumer/src/test/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateSinkFunctionConfigurationTests.java index 808396a0..a3bdcf3e 100644 --- a/functions/consumer/twitter-consumer/src/test/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateSinkFunctionConfigurationTests.java +++ b/functions/consumer/twitter-consumer/src/test/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateSinkFunctionConfigurationTests.java @@ -53,7 +53,7 @@ public class TwitterUpdateSinkFunctionConfigurationTests { } @Test - public void testToStatusUpdateQueryFunction() { + public void testMessageToStatusUpdateFunction() { TwitterUpdateConsumerProperties properties = new TwitterUpdateConsumerProperties(); properties.setAttachmentUrl(expression("'attachmentUrl'")); @@ -64,10 +64,10 @@ public class TwitterUpdateSinkFunctionConfigurationTests { properties.getLocation().setLat(expression("'37.78217'")); properties.getLocation().setLon(expression("'-122.40062'")); - Function, StatusUpdate> toStatusUpdateQueryFunction = - new TwitterUpdateConsumerConfiguration().toStatusUpdateQuery(properties); + Function, StatusUpdate> messageToStatusUpdateFunction = + new TwitterUpdateConsumerConfiguration().messageToStatusUpdateFunction(properties); - StatusUpdate result = toStatusUpdateQueryFunction.apply(new GenericMessage<>("Hello World")); + StatusUpdate result = messageToStatusUpdateFunction.apply(new GenericMessage<>("Hello World")); assertThat(result).isNotNull(); assertThat(result.getStatus()).isEqualTo("Hello World"); diff --git a/functions/function/twitter-function/README.adoc b/functions/function/twitter-function/README.adoc index 14b464d4..9d2e19af 100644 --- a/functions/function/twitter-function/README.adoc +++ b/functions/function/twitter-function/README.adoc @@ -19,7 +19,7 @@ You can import the `TwitterTrendFunctionConfiguration` in a Spring Boot applicat `filterFunction` -You can use `Function, Message> trendOrTrendLocationsFunction` as a qualifier when injecting. +You can use `Function, Message> twitterTrendFunction` as a qualifier when injecting. Once injected, you can use the `apply` method of the `Function` to invoke it and get the result. diff --git a/functions/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionConfiguration.java b/functions/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionConfiguration.java index 61253b87..c82f023c 100644 --- a/functions/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionConfiguration.java +++ b/functions/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionConfiguration.java @@ -80,7 +80,7 @@ public class TwitterTrendFunctionConfiguration { } @Bean - public Function, Message> trendOrTrendLocationsFunction( + public Function, Message> twitterTrendFunction( Function> managedJson, Function, Trends> trend, TwitterTrendFunctionProperties properties, Function, List> closestOrAvailableTrends) { diff --git a/functions/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierTests.java b/functions/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierTests.java index 809b5b35..196d7dc4 100644 --- a/functions/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierTests.java +++ b/functions/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierTests.java @@ -38,6 +38,7 @@ import org.springframework.boot.SpringBootConfiguration; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.fn.common.twitter.TwitterConnectionProperties; +import org.springframework.cloud.fn.common.twitter.util.TwitterTestUtils; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Primary; @@ -45,6 +46,7 @@ import org.springframework.messaging.Message; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.TestPropertySource; import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.util.SocketUtils; import static org.assertj.core.api.Assertions.assertThat; import static org.mockserver.matchers.Times.exactly; @@ -69,7 +71,7 @@ public abstract class TwitterStreamSupplierTests { private static final String MOCK_SERVER_IP = "127.0.0.1"; - private static final Integer MOCK_SERVER_PORT = 1080; + private static final Integer MOCK_SERVER_PORT = SocketUtils.findAvailableTcpPort(); private static ClientAndServer mockServer; diff --git a/functions/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterTestUtils.java b/functions/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterTestUtils.java deleted file mode 100644 index 69305887..00000000 --- a/functions/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterTestUtils.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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.fn.supplier.twitter.status.stream; - - -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 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); - } - } - -}