SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Address twitter review suggestions

Browse Source
This commit is contained in:
Christian Tzolov
2020-06-24 09:58:26 +02:00
parent bd48ead4e6
commit 9792768d55
24 changed files with 46 additions and 454 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
applications/processor/twitter-trend-processor/pom.xml
View File

@@ -67,7 +67,7 @@
<type>processor</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.twitter.trend.TwitterTrendFunctionConfiguration.class</configClass>
<functionDefinition>trendOrTrendLocationsFunction</functionDefinition>
<functionDefinition>twitterTrendFunction</functionDefinition>
</generatedApp>
<dependencies>

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

@@ -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<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);
}
}
}

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

@@ -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",

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

@@ -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",

4
applications/sink/twitter-message-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/message/TwitterMessageSinkIntegrationTests.java
View File

@@ -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;

68
applications/sink/twitter-message-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/message/TwitterTestUtils.java
View File

@@ -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<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);
}
}
}

4
applications/sink/twitter-update-sink/README.adoc
View File

@@ -32,13 +32,13 @@ $$twitter.update.text$$:: $$(SpEL expression) The text of the text update. URL e
`TwitterUpdateConsumerConfiguration` exposes 2 composable functions:
* `Function<Message<?>, StatusUpdate> toStatusUpdateQuery(TwitterUpdateConsumerProperties updateProperties)` - Converts input message into `StatusUpdate` query object.
* `Function<Message<?>, StatusUpdate> messageToStatusUpdateFunction(TwitterUpdateConsumerProperties updateProperties)` - Converts input message into `StatusUpdate` query object.
* `Consumer<StatusUpdate> 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

4
applications/sink/twitter-update-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/update/TwitterUpdateSinkIntegrationTests.java
View File

@@ -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;

4
applications/source/twitter-message-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/message/TwitterMessageSourceIntegrationTests.java
View File

@@ -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;

68
applications/source/twitter-message-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/message/TwitterTestUtils.java
View File

@@ -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<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);
}
}
}

4
applications/source/twitter-search-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/search/TwitterSearchSourceIntegrationTests.java
View File

@@ -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;

68
applications/source/twitter-search-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/search/TwitterTestUtils.java
View File

@@ -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<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);
}
}
}

4
applications/source/twitter-stream-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/stream/TwitterStreamSourceTests.java
View File

@@ -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;

68
applications/source/twitter-stream-source/src/test/java/org/springframework/cloud/stream/app/source/twitter/stream/TwitterTestUtils.java
View File

@@ -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<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);
}
}
}

3
functions/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionConfiguration.java
View File

@@ -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

2
applications/sink/twitter-update-sink/src/test/java/org/springframework/cloud/stream/app/sink/twitter/update/TwitterTestUtils.java → functions/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/util/TwitterTestUtils.java
View File

@@ -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;

4
functions/consumer/twitter-consumer/README.adoc
View File

@@ -22,9 +22,9 @@ You can import `TwitterUpdateConsumerConfiguration` in the application and then
- `Consumer<StatusUpdate> updateStatus` - if you have an `StatusUpdate` instance you can use the `updateStatus` to apply it.
- `Function<Message<?>, StatusUpdate> toStatusUpdateQuery` - function that converts a `Message<?>` text into a `StatusUpdate` instance using the `TwitterUpdateConsumerProperties` properties.
- `Function<Message<?>, StatusUpdate> messageToStatusUpdateFunction` - function that converts a `Message<?>` text into a `StatusUpdate` instance using the `TwitterUpdateConsumerProperties` properties.
- `Consumer<Message<?>> twitterStatusUpdateConsumer` - composes `toStatusUpdateQuery` and `updateStatus` to update the twitter status from Message text.
- `Consumer<Message<?>> 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`.

17
functions/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerConfiguration.java
View File

@@ -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<StatusUpdate> updateStatus(Twitter twitter) {
return statusUpdate -> {
@@ -74,7 +62,8 @@ public class TwitterUpdateConsumerConfiguration {
}
@Bean
public Function<Message<?>, StatusUpdate> toStatusUpdateQuery(TwitterUpdateConsumerProperties updateProperties) {
public Function<Message<?>, 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));

2
functions/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerProperties.java
View File

@@ -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;

8
functions/consumer/twitter-consumer/src/test/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateSinkFunctionConfigurationTests.java
View File

@@ -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<Message<?>, StatusUpdate> toStatusUpdateQueryFunction =
new TwitterUpdateConsumerConfiguration().toStatusUpdateQuery(properties);
Function<Message<?>, 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");

2
functions/function/twitter-function/README.adoc
View File

@@ -19,7 +19,7 @@ You can import the `TwitterTrendFunctionConfiguration` in a Spring Boot applicat
`filterFunction`
You can use `Function<Message<?>, Message<byte[]>> trendOrTrendLocationsFunction` as a qualifier when injecting.
You can use `Function<Message<?>, Message<byte[]>> twitterTrendFunction` as a qualifier when injecting.
Once injected, you can use the `apply` method of the `Function` to invoke it and get the result.

2
functions/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionConfiguration.java
View File

@@ -80,7 +80,7 @@ public class TwitterTrendFunctionConfiguration {
}
@Bean
public Function<Message<?>, Message<byte[]>> trendOrTrendLocationsFunction(
public Function<Message<?>, Message<byte[]>> twitterTrendFunction(
Function<Object, Message<byte[]>> managedJson, Function<Message<?>, Trends> trend,
TwitterTrendFunctionProperties properties, Function<Message<?>,
List<Location>> closestOrAvailableTrends) {

4
functions/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierTests.java
View File

@@ -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;

68
functions/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterTestUtils.java
View File

@@ -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<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);
}
}
}