diff --git a/common/twitter-common/pom.xml b/common/twitter-common/pom.xml new file mode 100644 index 00000000..d9d2a5fd --- /dev/null +++ b/common/twitter-common/pom.xml @@ -0,0 +1,56 @@ + + + 4.0.0 + twitter-common + 1.0.0-SNAPSHOT + twitter-common + Twitter common + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + 4.0.7 + + + + + org.twitter4j + twitter4j-stream + ${twitter4j.version} + + + + org.springframework.boot + spring-boot-starter-json + + + org.springframework.boot + spring-boot-starter-validation + + + + org.springframework.integration + spring-integration-ip + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + + diff --git a/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/Cursor.java b/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/Cursor.java new file mode 100644 index 00000000..ba6de357 --- /dev/null +++ b/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/Cursor.java @@ -0,0 +1,37 @@ +/* + * Copyright 2015-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.common.twitter; + +/** + * @author Christian Tzolov + */ +public class Cursor { + private long cursor = -1; + + public long getCursor() { + return cursor; + } + + public void updateCursor(long newCursor) { + this.cursor = (newCursor > 0) ? newCursor : -1; + } + + @Override + public String toString() { + return "Cursor{cursor=" + cursor + '}'; + } +} diff --git a/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/OnMissingStreamFunctionDefinitionCondition.java b/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/OnMissingStreamFunctionDefinitionCondition.java new file mode 100644 index 00000000..55565bdd --- /dev/null +++ b/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/OnMissingStreamFunctionDefinitionCondition.java @@ -0,0 +1,34 @@ +/* + * Copyright 2015-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.common.twitter; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.autoconfigure.condition.NoneNestedConditions; + +/** + * @author Christian Tzolov + */ +public class OnMissingStreamFunctionDefinitionCondition extends NoneNestedConditions { + + public OnMissingStreamFunctionDefinitionCondition() { + super(ConfigurationPhase.REGISTER_BEAN); + } + + @ConditionalOnProperty(name = "spring.cloud.stream.function.definition") + static class OnFunctionDslProperty { + } +} diff --git a/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionConfiguration.java b/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionConfiguration.java new file mode 100644 index 00000000..057204d2 --- /dev/null +++ b/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionConfiguration.java @@ -0,0 +1,130 @@ +/* + * Copyright 2015-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.common.twitter; + +import java.util.ArrayList; +import java.util.List; +import java.util.function.Function; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import twitter4j.Twitter; +import twitter4j.TwitterFactory; +import twitter4j.TwitterObjectFactory; +import twitter4j.TwitterStream; +import twitter4j.TwitterStreamFactory; +import twitter4j.conf.ConfigurationBuilder; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.Message; +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 + */ +@Configuration +@EnableConfigurationProperties({ TwitterConnectionProperties.class }) +public class TwitterConnectionConfiguration { + + private static final Log logger = LogFactory.getLog(TwitterConnectionConfiguration.class); + + @Bean + public twitter4j.conf.Configuration twitterConfiguration(TwitterConnectionProperties properties, + Function toConfigurationBuilder) { + return toConfigurationBuilder.apply(properties).build(); + } + + @Bean + public Twitter twitter(twitter4j.conf.Configuration configuration) { + return new TwitterFactory(configuration).getInstance(); + } + + @Bean + public TwitterStream twitterStream(twitter4j.conf.Configuration configuration) { + return new TwitterStreamFactory(configuration).getInstance(); + } + + @Bean + public Function toConfigurationBuilder() { + return properties -> new ConfigurationBuilder() + .setJSONStoreEnabled(properties.isRawJson()) + .setDebugEnabled(properties.isDebugEnabled()) + .setOAuthConsumerKey(properties.getConsumerKey()) + .setOAuthConsumerSecret(properties.getConsumerSecret()) + .setOAuthAccessToken(properties.getAccessToken()) + .setOAuthAccessTokenSecret(properties.getAccessTokenSecret()); + } + + @Bean + public Function> json(ObjectMapper mapper) { + return objects -> { + try { + String json = mapper.writeValueAsString(objects); + + return MessageBuilder + .withPayload(json.getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE) + .build(); + } + catch (JsonProcessingException e) { + logger.error("Status to JSON conversion error!", e); + } + return null; + }; + } + + /** + * Retrieves the raw JSON form of the provided object. + * + * Note that raw JSON forms can be retrieved only from the same thread invoked the last method + * call and will become inaccessible once another method call. + * + * @return Function that can retrieve the raw JSON object from the objects returned by the Twitter4J's APIs. + */ + @Bean + public Function rawJsonExtractor() { + return response -> { + if (response instanceof List) { + List responses = (List) response; + List rawJsonList = new ArrayList<>(); + for (Object object : responses) { + rawJsonList.add(TwitterObjectFactory.getRawJSON(object)); + } + return rawJsonList; + } + else { + return TwitterObjectFactory.getRawJSON(response); + } + }; + } + + @Bean + public Function> managedJson(TwitterConnectionProperties properties, + Function rawJsonExtractor, Function> json) { + return list -> (properties.isRawJson()) ? rawJsonExtractor.andThen(json).apply(list) : json.apply(list); + } +} diff --git a/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionProperties.java b/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionProperties.java new file mode 100644 index 00000000..a340277c --- /dev/null +++ b/common/twitter-common/src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionProperties.java @@ -0,0 +1,115 @@ +/* + * Copyright 2015-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.common.twitter; + +import javax.validation.constraints.NotEmpty; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.validation.annotation.Validated; + + +/** + * @author Christian Tzolov + */ +@ConfigurationProperties("twitter.connection") +@Validated +public class TwitterConnectionProperties { + + /** + * Your Twitter key. + */ + @NotEmpty + private String consumerKey; + + /** + * Your Twitter secret. + */ + @NotEmpty + private String consumerSecret; + + /** + * Your Twitter token. + */ + @NotEmpty + private String accessToken; + + /** + * Your Twitter token secret. + */ + @NotEmpty + private String accessTokenSecret; + + /** + * Enables Twitter4J debug mode. + */ + private boolean debugEnabled = false; + + /** + * Enable caching the original (raw) JSON objects as returned by the Twitter APIs. + * When set to False the result will use the Twitter4J's json representations. + * When set to True the result will use the original Twitter APISs json representations. + */ + private boolean rawJson = true; + + public String getConsumerKey() { + return consumerKey; + } + + public void setConsumerKey(String consumerKey) { + this.consumerKey = consumerKey; + } + + public String getConsumerSecret() { + return consumerSecret; + } + + public void setConsumerSecret(String consumerSecret) { + this.consumerSecret = consumerSecret; + } + + public String getAccessToken() { + return accessToken; + } + + public void setAccessToken(String accessToken) { + this.accessToken = accessToken; + } + + public String getAccessTokenSecret() { + return accessTokenSecret; + } + + public void setAccessTokenSecret(String accessTokenSecret) { + this.accessTokenSecret = accessTokenSecret; + } + + public boolean isDebugEnabled() { + return debugEnabled; + } + + public void setDebugEnabled(boolean debugEnabled) { + this.debugEnabled = debugEnabled; + } + + public boolean isRawJson() { + return this.rawJson; + } + + public void setRawJson(boolean rawJson) { + this.rawJson = rawJson; + } +} diff --git a/consumer/twitter-consumer/README.adoc b/consumer/twitter-consumer/README.adoc new file mode 100644 index 00000000..2e95ceac --- /dev/null +++ b/consumer/twitter-consumer/README.adoc @@ -0,0 +1,110 @@ +# Twitter Consumers + + +## 1. Twitter Status Update Consumer. + +Updates the authenticating user's current text (e.g Tweeting). + +NOTE: For each update attempt, the update text is compared with the authenticating user's recent Tweets. +Any attempt that would result in duplication will be blocked, resulting in a 403 error. +A user cannot submit the same text twice in a row. + +While not rate limited by the API, a user is limited in the number of Tweets they can create at a time. +The update limit for standard API is 300 in 3 hours windows. +If the number of updates posted by the user reaches the current allowed limit this method will return an HTTP 403 error. + +You can find details for the Update API here: https://developer.twitter.com/en/docs/tweets/post-and-engage/api-reference/post-statuses-update + + +### 1.1 Beans for injection + +You can import `TwitterUpdateConsumerConfiguration` in the application and then inject the following beans. + +- `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. + +- `Consumer> twitterStatusUpdateConsumer` - composes `toStatusUpdateQuery` 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`. + +You can use `twitterStatusUpdateConsumer` as a qualifier when injecting. + +### 1.2 Configuration Options + +All configuration properties are prefixed with `twitter.update`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerProperties.java[TwitterUpdateConsumerProperties]. + +The twitter function makes uses of link:../spel-function/README.adoc[SpEL function]. + +### 1.3 Other usage + +See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/twitter-update-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application where it makes a Twitter Update sink. + +## 2. Twitter Message Consumer. + +Send Direct Messages to a specified user from the authenticating user. +Requires a JSON POST body and `Content-Type` header to be set to `application/json`. + +NOTE: When a message is received from a user you may send up to 5 messages in response within a 24 hour window. +Each message received resets the 24 hour window and the 5 allotted messages. +Sending a 6th message within a 24 hour window or sending a message outside of a 24 hour window will count towards rate-limiting. +This behavior only applies when using the POST direct_messages/events/new endpoint. + +SpEL expressions are used to compute the request parameters from the input message. + +### 2.1 Beans for injection + +You can import `TwitterMessageConsumerConfiguration` in the application and then inject the following bean. + +- `Consumer> sendDirectMessageConsumer` + +Note: the Message content is expected to be in text format. Consider using the `byteArrayTextToString` utility `Function`. + +You can use `twitterStatusUpdateConsumer` as a qualifier when injecting. + +### 2.2 Configuration Options + +All configuration properties are prefixed with `twitter.message.update`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/twitter/message/TwitterMessageConsumerProperties.java[TwitterMessageConsumerProperties]. + +The twitter function makes uses of link:../spel-function/README.adoc[SpEL function]. + +### 2.3 Other usage + +See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/twitter-message-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application where it makes a Twitter Message sink. + +## 3. Twitter Friendship Consumer. + +Allows creating `follow`, `unfollow` and `update` relationships with specified `userId` or `screenName`. +The `twitter.friendships.sink.type` property allows to select the desired friendship operation. + +* https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/post-friendships-create[Friendships Create API] - Allows the authenticating user to follow (friend) the user specified in the ID parameter. +Actions taken in this method are asynchronous. +Changes will be eventually consistent. +* https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/post-friendships-update[Friendships Update API] - Enable or disable Retweets and device notifications from the specified user. +* https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/post-friendships-destroy[Friendships Destroy API] - Allows the authenticating user to unfollow the user specified in the ID parameter. + +SpEL expressions are used to compute the request parameters from the input message. +Every operation type has its own parameters. + +### 3.1 Beans for injection + +You can import `TwitterFriendshipsConsumerConfiguration` in the application and then inject the following bean. + +- `Consumer> friendshipConsumer` + +Note: the Message content is expected to be in text format. Consider using the `byteArrayTextToString` utility `Function`. + +You can use `friendshipConsumer` as a qualifier when injecting. + +### 3.2 Configuration Options + +All configuration properties are prefixed with `twitter.friendships.update`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/twitter/friendship/TwitterFriendshipsConsumerProperties.java[TwitterFriendshipsConsumerProperties]. + +The twitter function makes uses of link:../spel-function/README.adoc[SpEL function]. + diff --git a/consumer/twitter-consumer/pom.xml b/consumer/twitter-consumer/pom.xml new file mode 100644 index 00000000..678c5420 --- /dev/null +++ b/consumer/twitter-consumer/pom.xml @@ -0,0 +1,57 @@ + + + 4.0.0 + twitter-consumer + 1.0.0-SNAPSHOT + twitter-consumer + twitter consumer + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.cloud.fn + twitter-common + ${project.version} + + + org.springframework.cloud.fn + payload-converter-function + ${project.version} + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + org.springframework.integration + spring-integration-test + test + + + + diff --git a/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/friendship/TwitterFriendshipsConsumerConfiguration.java b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/friendship/TwitterFriendshipsConsumerConfiguration.java new file mode 100644 index 00000000..ae55a818 --- /dev/null +++ b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/friendship/TwitterFriendshipsConsumerConfiguration.java @@ -0,0 +1,96 @@ +/* + * 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.consumer.twitter.friendship; + +import java.util.function.Consumer; + +import twitter4j.Twitter; +import twitter4j.TwitterException; + +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.messaging.Message; + +/** + * + * @author Christian Tzolov + */ +@Configuration +@EnableConfigurationProperties(TwitterFriendshipsConsumerProperties.class) +@Import(TwitterConnectionConfiguration.class) +public class TwitterFriendshipsConsumerConfiguration { + + @Bean + @SuppressWarnings("Duplicates") + public Consumer> friendshipConsumer(TwitterFriendshipsConsumerProperties properties, Twitter twitter) { + + return message -> { + try { + TwitterFriendshipsConsumerProperties.OperationType type = + properties.getType().getValue(message, TwitterFriendshipsConsumerProperties.OperationType.class); + //TwitterFriendshipsSinkProperties.OperationType type = TwitterFriendshipsSinkProperties.OperationType.create; + if (properties.getUserId() != null) { + Long userId = properties.getUserId().getValue(message, long.class); + switch (type) { + case create: + boolean follow = properties.getCreate().getFollow().getValue(message, boolean.class); + twitter.createFriendship(userId, follow); + return; + + case update: + boolean enableDeviceNotification = properties.getUpdate().getDevice().getValue(message, boolean.class); + boolean retweets = properties.getUpdate().getRetweets().getValue(message, boolean.class); + twitter.updateFriendship(userId, enableDeviceNotification, retweets); + return; + + case destroy: + twitter.destroyFriendship(userId); + return; + } + } + else if (properties.getScreenName() != null) { + String screenName = properties.getScreenName().getValue(message, String.class); + switch (type) { + case create: + boolean follow = properties.getCreate().getFollow().getValue(message, boolean.class); + twitter.createFriendship(screenName, follow); + return; + + case update: + boolean enableDeviceNotification = properties.getUpdate().getDevice().getValue(message, boolean.class); + boolean retweets = properties.getUpdate().getRetweets().getValue(message, boolean.class); + twitter.updateFriendship(screenName, enableDeviceNotification, retweets); + return; + + case destroy: + twitter.destroyFriendship(screenName); + return; + } + } + else { + throw new IllegalStateException("Either ScreenName or UserID must be set"); + } + } + catch (TwitterException te) { + throw new IllegalStateException("Twitter API error!", te); + } + }; + } +} diff --git a/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/friendship/TwitterFriendshipsConsumerProperties.java b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/friendship/TwitterFriendshipsConsumerProperties.java new file mode 100644 index 00000000..981585b4 --- /dev/null +++ b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/friendship/TwitterFriendshipsConsumerProperties.java @@ -0,0 +1,149 @@ +/* + * 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.consumer.twitter.friendship; + +import javax.validation.constraints.AssertTrue; +import javax.validation.constraints.NotNull; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.stereotype.Component; +import org.springframework.validation.annotation.Validated; + + +/** + * @author Christian Tzolov + */ +@Component +@ConfigurationProperties("twitter.friendships.update") +@Validated +public class TwitterFriendshipsConsumerProperties { + + public enum OperationType { + /** Friendship operation types. */ + create, update, destroy + } + + /** + * The screen name of the user to follow (String). + */ + private Expression screenName; + + /** + * The ID of the user to follow (Integer). + */ + private Expression userId; + + /** + * Type of Friendships request. + */ + private Expression type = new SpelExpressionParser().parseExpression("'create'"); + + /** + * Additional properties for the Friendships create requests. + */ + private Create create = new Create(); + + /** + * Additional properties for the Friendships update requests. + */ + private Update update = new Update(); + + public Expression getScreenName() { + return screenName; + } + + public void setScreenName(Expression screenName) { + this.screenName = screenName; + } + + public Expression getUserId() { + return userId; + } + + public void setUserId(Expression userId) { + this.userId = userId; + } + + public Expression getType() { + return type; + } + + public void setType(Expression type) { + this.type = type; + } + + public Create getCreate() { + return create; + } + + public Update getUpdate() { + return update; + } + + @AssertTrue(message = "Either userId or screenName must be provided") + public boolean isUserProvided() { + return this.userId != null || this.screenName != null; + } + + public static class Create { + /** + * The ID of the user to follow (boolean). + */ + @NotNull + private Expression follow = new SpelExpressionParser().parseExpression("'true'"); + + public Expression getFollow() { + return follow; + } + + public void setFollow(Expression follow) { + this.follow = follow; + } + } + + public static class Update { + /** + * Enable/disable device notifications from the target user. + */ + @NotNull + private Expression device = new SpelExpressionParser().parseExpression("'true'"); + + /** + * Enable/disable Retweets from the target user. + */ + @NotNull + private Expression retweets = new SpelExpressionParser().parseExpression("'true'"); + + public Expression getDevice() { + return device; + } + + public void setDevice(Expression device) { + this.device = device; + } + + public Expression getRetweets() { + return retweets; + } + + public void setRetweets(Expression retweets) { + this.retweets = retweets; + } + } +} diff --git a/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/message/TwitterMessageConsumerConfiguration.java b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/message/TwitterMessageConsumerConfiguration.java new file mode 100644 index 00000000..377d8319 --- /dev/null +++ b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/message/TwitterMessageConsumerConfiguration.java @@ -0,0 +1,71 @@ +/* + * 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.consumer.twitter.message; + +import java.util.function.Consumer; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import twitter4j.Twitter; +import twitter4j.TwitterException; + +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.messaging.Message; + +/** + * + * @author Christian Tzolov + */ +@Configuration +@EnableConfigurationProperties(TwitterMessageConsumerProperties.class) +@Import(TwitterConnectionConfiguration.class) +public class TwitterMessageConsumerConfiguration { + + private static final Log logger = LogFactory.getLog(TwitterMessageConsumerConfiguration.class); + + @Bean + public Consumer> sendDirectMessageConsumer(TwitterMessageConsumerProperties messageProperties, Twitter twitter) { + return message -> { + try { + String messageText = messageProperties.getText().getValue(message, String.class); + + if (messageProperties.getUserId() != null) { + Long userId = messageProperties.getUserId().getValue(message, long.class); + if (messageProperties.getMediaId() != null) { + Long mediaId = messageProperties.getMediaId().getValue(message, long.class); + twitter.sendDirectMessage(userId, messageText, mediaId); + } + twitter.sendDirectMessage(userId, messageText); + } + else if (messageProperties.getScreenName() != null) { + String screenName = messageProperties.getScreenName().getValue(message, String.class); + twitter.sendDirectMessage(screenName, messageText); + } + else { + throw new RuntimeException("Either the UserId or screenName must be set"); + } + } + catch (TwitterException e) { + logger.error("Failed to process message:" + message, e); + } + }; + } +} diff --git a/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/message/TwitterMessageConsumerProperties.java b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/message/TwitterMessageConsumerProperties.java new file mode 100644 index 00000000..e37be02b --- /dev/null +++ b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/message/TwitterMessageConsumerProperties.java @@ -0,0 +1,85 @@ +/* + * 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.consumer.twitter.message; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.validation.annotation.Validated; + + +/** + * @author Christian Tzolov + */ +@ConfigurationProperties("twitter.message.update") +@Validated +public class TwitterMessageConsumerProperties { + + private static final Expression DEFAULT_EXPRESSION = new SpelExpressionParser().parseExpression("payload"); + + /** + * The direct message text. URL encode as necessary. Max length of 10,000 characters. + */ + private Expression text = DEFAULT_EXPRESSION; + + /** + * The screen name of the user to whom send the direct message. + */ + private Expression screenName; + + /** + * The user id of the user to whom send the direct message. + */ + private Expression userId; + + /** + * A media id to associate with the message. A Direct Message may only reference a single media id. + */ + private Expression mediaId; + + public Expression getUserId() { + return userId; + } + + public void setUserId(Expression userId) { + this.userId = userId; + } + + public void setText(Expression text) { + this.text = text; + } + + public Expression getText() { + return text; + } + + public Expression getScreenName() { + return screenName; + } + + public void setScreenName(Expression screenName) { + this.screenName = screenName; + } + + public Expression getMediaId() { + return mediaId; + } + + public void setMediaId(Expression mediaId) { + this.mediaId = mediaId; + } +} diff --git a/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerConfiguration.java b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerConfiguration.java new file mode 100644 index 00000000..cb301826 --- /dev/null +++ b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerConfiguration.java @@ -0,0 +1,125 @@ +/* + * 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.consumer.twitter.status.update; + +import java.util.Properties; +import java.util.function.Consumer; +import java.util.function.Function; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import twitter4j.GeoLocation; +import twitter4j.Status; +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; + +/** + * + * @author Christian Tzolov + */ +@Configuration +@EnableConfigurationProperties(TwitterUpdateConsumerProperties.class) +@Import(TwitterConnectionConfiguration.class) +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 -> { + try { + Status status = twitter.updateStatus(statusUpdate); + + if (logger.isDebugEnabled()) { + logger.debug(status); + } + } + catch (TwitterException e) { + logger.error("Failed apply update status: " + statusUpdate, e); + } + }; + } + + @Bean + public Function, StatusUpdate> toStatusUpdateQuery(TwitterUpdateConsumerProperties updateProperties) { + + return message -> { + + String updateText = updateProperties.getText().getValue(message, String.class); + + StatusUpdate statusUpdate = new StatusUpdate(updateText); + + if (updateProperties.getAttachmentUrl() != null) { + statusUpdate.setAttachmentUrl(updateProperties.getAttachmentUrl().getValue(message, String.class)); + } + + if (updateProperties.getPlaceId() != null) { + statusUpdate.setPlaceId(updateProperties.getPlaceId().getValue(message, String.class)); + } + + if (updateProperties.getInReplyToStatusId() != null) { + statusUpdate.setInReplyToStatusId(updateProperties.getInReplyToStatusId().getValue(message, int.class)); + statusUpdate.setAutoPopulateReplyMetadata(true); + } + + if (updateProperties.getDisplayCoordinates() != null) { + statusUpdate.setDisplayCoordinates( + updateProperties.getDisplayCoordinates().getValue(message, boolean.class)); + } + + if (updateProperties.getMediaIds() != null) { + long[] mediaIds = updateProperties.getMediaIds().getValue(message, long[].class); + statusUpdate.setMediaIds(mediaIds); + } + + 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)); + } + + return statusUpdate; + }; + } + + @Bean + public Consumer> twitterStatusUpdateConsumer(Function, StatusUpdate> statusUpdateQuery, + Consumer updateStatus) { + return message -> updateStatus.accept(statusUpdateQuery.apply(message)); + } +} diff --git a/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerProperties.java b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerProperties.java new file mode 100644 index 00000000..5ce35d7f --- /dev/null +++ b/consumer/twitter-consumer/src/main/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateConsumerProperties.java @@ -0,0 +1,174 @@ +/* + * 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.consumer.twitter.status.update; + +import javax.validation.constraints.AssertTrue; +import javax.validation.constraints.NotNull; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.validation.annotation.Validated; + + +/** + * @author Christian Tzolov + */ +@ConfigurationProperties("twitter.update") +@Validated +public class TwitterUpdateConsumerProperties { + + private static final Expression DEFAULT_EXPRESSION = new SpelExpressionParser().parseExpression("payload"); + + /** + * (SpEL expression) The text of the text update. URL encode as necessary. t.co link wrapping will + * affect character counts. Defaults to message's payload + */ + @NotNull + private Expression text = DEFAULT_EXPRESSION; + + /** + * (SpEL expression) In order for a URL to not be counted in the text body of an extended Tweet, provide a URL as a Tweet attachment. + * This URL must be a Tweet permalink, or Direct Message deep link. Arbitrary, non-Twitter URLs must remain in + * the text text. URLs passed to the attachment_url parameter not matching either a Tweet permalink or Direct + * Message deep link will fail at Tweet creation and cause an exception. + */ + private Expression attachmentUrl; + + /** + * (SpEL expression) A place in the world. + */ + private Expression placeId; + + /** + * (SpEL expression) The ID of an existing text that the update is in reply to. Note: This parameter will be ignored unless the + * author of the Tweet this parameter references is mentioned within the text text. Therefore, you must + * include @username, where username is the author of the referenced Tweet, within the update. + * + * When inReplyToStatusId is set the auto_populate_reply_metadata is automatically set as well. Later ensures + * that leading @mentions will be looked up from the original Tweet, and added to the new Tweet from there. + * This wil append @mentions into the metadata of an extended Tweet as a reply chain grows, until the limit + * on @mentions is reached. In cases where the original Tweet has been deleted, + * the reply will fail. + */ + private Expression inReplyToStatusId; + + /** + * (SpEL expression) Whether or not to put a pin on the exact coordinates a Tweet has been sent from. + */ + private Expression displayCoordinates; + + /** + * (SpEL expression) A comma-delimited list of media_ids to associate with the Tweet. You may include up to 4 photos or 1 animated + * GIF or 1 video in a Tweet. See Uploading Media for further details on uploading media. + */ + private Expression mediaIds; + + /** + * (SpEL expression) The location this Tweet refers to. Ignored if geo_enabled for the user is false! + */ + private Location location = new Location(); + + public Expression getText() { + return text; + } + + public void setText(Expression text) { + this.text = text; + } + + public Expression getAttachmentUrl() { + return attachmentUrl; + } + + public void setAttachmentUrl(Expression attachmentUrl) { + this.attachmentUrl = attachmentUrl; + } + + public Expression getPlaceId() { + return placeId; + } + + public void setPlaceId(Expression placeId) { + this.placeId = placeId; + } + + public Expression getInReplyToStatusId() { + return inReplyToStatusId; + } + + public void setInReplyToStatusId(Expression inReplyToStatusId) { + this.inReplyToStatusId = inReplyToStatusId; + } + + public Expression getDisplayCoordinates() { + return displayCoordinates; + } + + public void setDisplayCoordinates(Expression displayCoordinates) { + this.displayCoordinates = displayCoordinates; + } + + public Expression getMediaIds() { + return mediaIds; + } + + public void setMediaIds(Expression mediaIds) { + this.mediaIds = mediaIds; + } + + public Location getLocation() { + return location; + } + + @AssertTrue(message = "Lat and Long must be set together or both not being set") + public boolean validateLatLon() { + return (this.getLocation().getLat() != null && this.getLocation().getLon() != null) + || (this.getLocation().getLat() == null && this.getLocation().getLon() == null); + } + + public static class Location { + /** + * The latitude of the location this Tweet refers to. This parameter will be ignored unless it is inside the range + * -90.0 to +90.0 (North is positive) inclusive. It will also be ignored if there is no corresponding long parameter. + */ + private Expression lat; + + /** + * The longitude of the location this Tweet refers to. The valid ranges for longitude are -180.0 to +180.0 (East + * is positive) inclusive. This parameter will be ignored if outside that range, if it is not a number, + * if geo_enabled is disabled, or if there no corresponding lat parameter. + */ + private Expression lon; + + public Expression getLat() { + return lat; + } + + public void setLat(Expression lat) { + this.lat = lat; + } + + public Expression getLon() { + return lon; + } + + public void setLon(Expression lon) { + this.lon = lon; + } + } +} diff --git a/consumer/twitter-consumer/src/test/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateSinkFunctionConfigurationTests.java b/consumer/twitter-consumer/src/test/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateSinkFunctionConfigurationTests.java new file mode 100644 index 00000000..808396a0 --- /dev/null +++ b/consumer/twitter-consumer/src/test/java/org/springframework/cloud/fn/consumer/twitter/status/update/TwitterUpdateSinkFunctionConfigurationTests.java @@ -0,0 +1,86 @@ +/* + * 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.consumer.twitter.status.update; + +import java.util.function.Consumer; +import java.util.function.Function; + +import org.junit.Test; +import twitter4j.StatusUpdate; +import twitter4j.Twitter; +import twitter4j.TwitterException; + +import org.springframework.expression.Expression; +import org.springframework.expression.ExpressionParser; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +/** + * @author Christian Tzolov + */ +public class TwitterUpdateSinkFunctionConfigurationTests { + + @Test + public void testStatusUpdateConsumer() throws TwitterException { + Twitter twitter = mock(Twitter.class); + + Consumer statusUpdateConsumer = + new TwitterUpdateConsumerConfiguration().updateStatus(twitter); + + StatusUpdate statusUpdateQuery = new StatusUpdate("Hello World"); + statusUpdateConsumer.accept(statusUpdateQuery); + verify(twitter).updateStatus(eq(statusUpdateQuery)); + } + + @Test + public void testToStatusUpdateQueryFunction() { + TwitterUpdateConsumerProperties properties = new TwitterUpdateConsumerProperties(); + + properties.setAttachmentUrl(expression("'attachmentUrl'")); + properties.setPlaceId(expression("'myPlaceId'")); + properties.setInReplyToStatusId(expression("'666666'")); + properties.setDisplayCoordinates(expression("'true'")); + properties.setMediaIds(expression("'471592142565957632, 471592142565957633'")); + properties.getLocation().setLat(expression("'37.78217'")); + properties.getLocation().setLon(expression("'-122.40062'")); + + Function, StatusUpdate> toStatusUpdateQueryFunction = + new TwitterUpdateConsumerConfiguration().toStatusUpdateQuery(properties); + + StatusUpdate result = toStatusUpdateQueryFunction.apply(new GenericMessage<>("Hello World")); + + assertThat(result).isNotNull(); + assertThat(result.getStatus()).isEqualTo("Hello World"); + assertThat(result.getAttachmentUrl()).isEqualTo("attachmentUrl"); + assertThat(result.getPlaceId()).isEqualTo("myPlaceId"); + assertThat(result.getInReplyToStatusId()).isEqualTo(666666L); + assertThat(result.isDisplayCoordinates()).isTrue(); + assertThat(result.getLocation().getLatitude()).isEqualTo(37.78217); + assertThat(result.getLocation().getLongitude()).isEqualTo(-122.40062); + } + + private Expression expression(String expressionString) { + ExpressionParser parser = new SpelExpressionParser(); + return parser.parseExpression(expressionString); + } +} diff --git a/function/twitter-function/README.adoc b/function/twitter-function/README.adoc new file mode 100644 index 00000000..14b464d4 --- /dev/null +++ b/function/twitter-function/README.adoc @@ -0,0 +1,30 @@ +# Twitter Functions + +This module provides couple of twitter functions that can be reused and composed in other applications. + +## Twitter Trend Function + +Functions can return either Trends topics or the Locations of the trending topics. The `twitter.trend.trend-query-type` property allows choosing between both types. + +* Trends - `twitter.trend.trend-query-type` is set to `trend`. Leverages the https://developer.twitter.com/en/docs/trends/trends-for-location/api-reference/get-trends-place[Trends API] to return the https://help.twitter.com/en/using-twitter/twitter-trending-faqs[trending topics] near a specific latitude, longitude location. + +* Trend Locations - the `twitter.trend.trend-query-type` is set `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. + +### Beans for injection + +You can import the `TwitterTrendFunctionConfiguration` in a Spring Boot application and then inject the following bean. + +`filterFunction` + +You can use `Function, Message> trendOrTrendLocationsFunction` as a qualifier when injecting. + +Once injected, you can use the `apply` method of the `Function` to invoke it and get the result. + +### Configuration Options + + +### Other usage + diff --git a/function/twitter-function/pom.xml b/function/twitter-function/pom.xml new file mode 100644 index 00000000..93098c54 --- /dev/null +++ b/function/twitter-function/pom.xml @@ -0,0 +1,46 @@ + + + 4.0.0 + twitter-function + 1.0.0-SNAPSHOT + twitter-function + twitter functions + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.cloud.fn + twitter-common + ${project.version} + + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + + diff --git a/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionConfiguration.java b/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionConfiguration.java new file mode 100644 index 00000000..61253b87 --- /dev/null +++ b/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionConfiguration.java @@ -0,0 +1,91 @@ +/* + * 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.twitter.trend; + +import java.util.List; +import java.util.function.Function; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import twitter4j.GeoLocation; +import twitter4j.Location; +import twitter4j.Trends; +import twitter4j.Twitter; +import twitter4j.TwitterException; + +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.messaging.Message; + +/** + * @author Christian Tzolov + */ +@Configuration +@EnableConfigurationProperties(TwitterTrendFunctionProperties.class) +@Import(TwitterConnectionConfiguration.class) +public class TwitterTrendFunctionConfiguration { + + private static final Log logger = LogFactory.getLog(TwitterTrendFunctionConfiguration.class); + + @Bean + public Function, Trends> trend(TwitterTrendFunctionProperties properties, Twitter twitter) { + return message -> { + try { + int woeid = properties.getLocationId().getValue(message, int.class); + return twitter.getPlaceTrends(woeid); + } + catch (TwitterException e) { + logger.error("Twitter API error!", e); + } + return null; + }; + } + + @Bean + public Function, List> closestOrAvailableTrends( + TwitterTrendFunctionProperties properties, Twitter twitter) { + return message -> { + try { + if (properties.getClosest().getLat() != null && properties.getClosest().getLon() != null) { + double lat = properties.getClosest().getLat().getValue(message, double.class); + double lon = properties.getClosest().getLon().getValue(message, double.class); + return twitter.getClosestTrends(new GeoLocation(lat, lon)); + } + else { + return twitter.getAvailableTrends(); + } + } + catch (TwitterException e) { + logger.error("Twitter API error!", e); + } + return null; + }; + } + + @Bean + public Function, Message> trendOrTrendLocationsFunction( + Function> managedJson, Function, Trends> trend, + TwitterTrendFunctionProperties properties, Function, + List> closestOrAvailableTrends) { + + return (properties.getTrendQueryType() == TwitterTrendFunctionProperties.TrendQueryType.trend) ? + trend.andThen(managedJson) : closestOrAvailableTrends.andThen(managedJson); + } +} diff --git a/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionProperties.java b/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionProperties.java new file mode 100644 index 00000000..95ec7704 --- /dev/null +++ b/function/twitter-function/src/main/java/org/springframework/cloud/fn/twitter/trend/TwitterTrendFunctionProperties.java @@ -0,0 +1,107 @@ +/* + * 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.twitter.trend; + +import javax.validation.constraints.NotNull; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.validation.annotation.Validated; + +/** + * @author Christian Tzolov + */ +@ConfigurationProperties("twitter.trend") +@Validated +public class TwitterTrendFunctionProperties { + + private static final Expression DEFAULT_EXPRESSION = new SpelExpressionParser().parseExpression("payload"); + + enum TrendQueryType { + /** Retrieve trending places. */ + trend, + /** Retrieve the Locations of trending places. */ + trendLocation + } + + private TrendQueryType trendQueryType = TrendQueryType.trend; + + public TrendQueryType getTrendQueryType() { + return trendQueryType; + } + + public void setTrendQueryType(TrendQueryType trendQueryType) { + this.trendQueryType = trendQueryType; + } + + /** + * The Yahoo! Where On Earth ID of the location to return trending information for. + * Global information is available by using 1 as the WOEID. + */ + @NotNull + private Expression locationId = DEFAULT_EXPRESSION; + + public Expression getLocationId() { + return locationId; + } + + public void setLocationId(Expression locationId) { + this.locationId = locationId; + } + + /** + * + */ + private Closest closest = new Closest(); + + public Closest getClosest() { + return closest; + } + + public static class Closest { + /** + * 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. + */ + private Expression lat; + + /** + * 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. + */ + private Expression lon; + + public Expression getLat() { + return lat; + } + + public void setLat(Expression lat) { + this.lat = lat; + } + + public Expression getLon() { + return lon; + } + + public void setLon(Expression lon) { + this.lon = lon; + } + } +} diff --git a/pom.xml b/pom.xml index 8d5983e7..2849499a 100644 --- a/pom.xml +++ b/pom.xml @@ -50,6 +50,7 @@ common/metadata-store-common common/mqtt-common common/tcp-common + common/twitter-common consumer/cassandra-consumer @@ -67,6 +68,7 @@ consumer/tcp-consumer consumer/websocket-consumer consumer/s3-consumer + consumer/twitter-consumer function/filter-function function/header-enricher-function @@ -76,6 +78,7 @@ function/splitter-function function/tasklauncher-function function/task-launch-request-function + function/twitter-function supplier/file-supplier supplier/ftp-supplier @@ -90,6 +93,7 @@ supplier/rabbit-supplier supplier/websocket-supplier supplier/s3-supplier + supplier/twitter-supplier spring-functions-parent diff --git a/supplier/twitter-supplier/README.adoc b/supplier/twitter-supplier/README.adoc new file mode 100644 index 00000000..9c795586 --- /dev/null +++ b/supplier/twitter-supplier/README.adoc @@ -0,0 +1,143 @@ +# Twitter Suppliers + +This module provides a Twitter Status, Message, Friendship suppliers that can be reused and composed in various applications. + +`java.util.function.Supplier` + + +## 1. Twitter Status Search + +The Twitter's https://developer.twitter.com/en/docs/tweets/search/api-reference/get-search-tweets.html[Standard search API] (search/tweets) allows simple queries against the indices of recent or popular Tweets. This `Source` provides continuous searches against a sampling of recent Tweets published in the past 7 days. Part of the 'public' set of APIs. + +Returns a collection of relevant Tweets matching a specified query. + +### 1.1 Beans for injection + +You can import the `TwitterSearchSupplierConfiguration` in the application and then inject the following bean. + +`twitterSearchSupplier` + +You need to inject this as `Supplier>`. + +You can use `twitterSearchSupplier` as a qualifier when injecting. + +Once injected, you can use the `get` method of the `Supplier` to invoke it. + +### 1.2 Configuration Options + +The configuration properties prefixed with `twitter.search`. +There are also properties that need to be used with the prefix `twitter.connection`. + +The `spring.cloud.stream.poller` properties control the interval between consecutive search requests. Rate Limit - 180 requests per 30 min. window (e.g. ~6 r/m, ~ 1 req / 10 sec.) + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/twitter/search/stream/TwitterSearchSupplierProperties.java[TwitterSearchSupplierProperties]. +See link:src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionProperties.java[TwitterConnectionProperties] and https://github.com/spring-cloud/spring-cloud-stream/blob/master/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DefaultPollerProperties.java[DefaultPollerProperties] + +### 1.3 Other usage + +See this https://github.com/spring-cloud/stream-applications/blob/master/applications/source/twitter-search-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application where it makes a Twitter Search Source. + +## 2. Twitter Status Real-time Retrieval + +Provides real-time, Tweet streaming based on the https://developer.twitter.com/en/docs/tweets/filter-realtime/api-reference/post-statuses-filter.html[Filter] and https://developer.twitter.com/en/docs/tweets/sample-realtime/overview/GET_statuse_sample[Sample] APIs. +The `Filter API` flavor returns public statuses that match one or more filter predicates. +The `Sample API` flavor returns a small random sample of all public statuses. + +This supplier gives you a reactive stream of tweets from the configured connection as the supplier has a signature of `Supplier>>`. +Users have to subscribe to this `Flux` and receive the data. + +The default access level allows up to 400 track keywords, 5,000 follow user Ids and 25 0.1-360 degree location boxes. + +### 2.1 Beans for injection + +You can import the `TwitterStreamSupplierConfiguration` in the application and then inject the following bean. + +`twitterStreamSupplier` + +You need to inject this as `Supplier>>`. + +You can use `twitterStreamSupplier` as a qualifier when injecting. + +Once injected, you can use the `get` method of the `Supplier` to invoke it and then subscribe to the returned `Flux`. + +### 2.2 Configuration Options + +All configuration properties are prefixed with `twitter.stream`. +There are also properties that need to be used with the prefix `twitter.connection`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierProperties.java[TwitterStreamSupplierProperties]. +See link:src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionProperties.java[TwitterConnectionProperties] also. + + +### 2.3 Other usage + +See this https://github.com/spring-cloud/stream-applications/blob/master/applications/source/twitter-stream-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application where it makes a Twitter Stream Source. + +## 3. Twitter Direct Message Supplier + +Repeatedly retrieves the direct messages (both sent and received) within the last 30 days, sorted in reverse-chronological order. +The relieved messages are cached (in a `MetadataStore` cache) to prevent duplications. +By default an in-memory `SimpleMetadataStore` is used. + +The `twitter.message.source.count` controls the number or returned messages. + +The `spring.cloud.stream.poller` properties control the message poll interval. +Must be aligned with used APIs rate limit + +### 3.1 Beans for injection + +You can import the `TwitterMessageSupplierConfiguration` in the application and then inject the following bean. + +`twitterMessageSupplier` + +You need to inject this as `Supplier>`. + +You can use `twitterMessageSupplier` as a qualifier when injecting. + +Once injected, you can use the `get` method of the `Supplier` to invoke it. + +### 3.2 Configuration Options + +The configuration properties prefixed with `twitter.search`. +There are also properties that need to be used with the prefix `twitter.connection`. + +The `spring.cloud.stream.poller` properties control the interval between consecutive search requests. Rate Limit - 180 requests per 30 min. window (e.g. ~6 r/m, ~ 1 req / 10 sec.) + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/twitter/message/TwitterMessageSupplierProperties.java[TwitterMessageSupplierProperties]. +See link:src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionProperties.java[TwitterConnectionProperties] and https://github.com/spring-cloud/spring-cloud-stream/blob/master/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DefaultPollerProperties.java[DefaultPollerProperties] + +### 3.3 Other usage + +See this https://github.com/spring-cloud/stream-applications/blob/master/applications/source/twitter-message-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application where it makes a Twitter Message Source. + +## 4. Twitter Friendships Supplier + +Returns a cursored collection of user objects either for the https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/get-followers-list[users following the specified user] (`followers`) or for https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/get-friends-list[every user the specified user is following] (`friends`). + +The `twitter.friendships.source.type` property allow to select between both types. + +TIP: Rate limit: 15 requests per 30 min window. ~ 1 req/ 2 min + +### 4.1 Beans for injection + +You can import the `TwitterFriendshipsSupplierConfiguration` in the application and then inject one the following beans. + +- `followersSupplier` (only if `twitter.friendships.source.type=followers` ) - retrieves the https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/get-followers-list[users following the specified user] (`followers`) + +- `friendsSupplier` (only if `twitter.friendships.source.type=friends`) - retrieves the for https://developer.twitter.com/en/docs/accounts-and-users/follow-search-get-users/api-reference/get-friends-list[every user the specified user is following] (`friends`). + +Both suppliers expose `Supplier>`. + +- `deduplicatedFriendsJsonSupplier` - retrieves either the followers, or the friends collection (controlled by the `twitter.friendships.source.type`) property, . +encoded as JSON `Message` payloads. You need to inject this as `Supplier>`. + + +### 4.2 Configuration Options + +The configuration properties prefixed with `twitter.friendships.source`. +There are also properties that need to be used with the prefix `twitter.connection`. + +The `spring.cloud.stream.poller` properties control the interval between consecutive search requests. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/twitter/friendships/TwitterFriendshipsSupplierProperties.java[TwitterFriendshipsSupplierProperties]. +See link:src/main/java/org/springframework/cloud/fn/common/twitter/TwitterConnectionProperties.java[TwitterConnectionProperties] and https://github.com/spring-cloud/spring-cloud-stream/blob/master/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/DefaultPollerProperties.java[DefaultPollerProperties] diff --git a/supplier/twitter-supplier/pom.xml b/supplier/twitter-supplier/pom.xml new file mode 100644 index 00000000..b9fd36c1 --- /dev/null +++ b/supplier/twitter-supplier/pom.xml @@ -0,0 +1,80 @@ + + + 4.0.0 + twitter-supplier + 1.0.0-SNAPSHOT + twitter-supplier + twitter suppliers + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.cloud.fn + twitter-common + ${project.version} + + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.integration + spring-integration-jms + + + javax.jms + javax.jms-api + provided + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.apache.activemq + activemq-broker + test + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + org.springframework.integration + spring-integration-test + test + + + + org.mock-server + mockserver-netty + 5.10 + test + + + org.mock-server + mockserver-client-java + 5.10 + test + + + + diff --git a/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/friendships/TwitterFriendshipsSupplierConfiguration.java b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/friendships/TwitterFriendshipsSupplierConfiguration.java new file mode 100644 index 00000000..58587056 --- /dev/null +++ b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/friendships/TwitterFriendshipsSupplierConfiguration.java @@ -0,0 +1,150 @@ +/* + * 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.friendships; + +import java.util.ArrayList; +import java.util.List; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import twitter4j.PagableResponseList; +import twitter4j.Twitter; +import twitter4j.TwitterException; +import twitter4j.User; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.twitter.Cursor; +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.integration.metadata.MetadataStore; +import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.messaging.Message; + +/** + * + * @author Christian Tzolov + */ +@Configuration +@EnableConfigurationProperties(TwitterFriendshipsSupplierProperties.class) +@Import(TwitterConnectionConfiguration.class) +public class TwitterFriendshipsSupplierConfiguration { + + private static final Log logger = LogFactory.getLog(TwitterFriendshipsSupplierConfiguration.class); + + @Bean + @ConditionalOnMissingBean + public MetadataStore metadataStore() { + return new SimpleMetadataStore(); + } + + @Bean + @ConditionalOnMissingBean + public Cursor cursor() { + return new Cursor(); + } + + + @Bean + @ConditionalOnProperty(name = "twitter.friendships.source.type", havingValue = "followers") + public Supplier> followersSupplier(TwitterFriendshipsSupplierProperties properties, + Twitter twitter, Cursor cursorState) { + return () -> { + try { + PagableResponseList users; + if (properties.getUserId() != null) { + users = twitter.getFollowersList(properties.getUserId(), cursorState.getCursor(), + properties.getCount(), properties.isSkipStatus(), properties.isIncludeUserEntities()); + } + else { // by ScreenName + users = twitter.getFollowersList(properties.getScreenName(), cursorState.getCursor(), + properties.getCount(), properties.isSkipStatus(), properties.isIncludeUserEntities()); + } + + if (users != null) { + cursorState.updateCursor(users.getNextCursor()); + return users; + } + + logger.error(String.format("NULL users response for properties: %s and cursor: %s!", properties, cursorState)); + cursorState.updateCursor(-1); + } + catch (TwitterException e) { + logger.error("Twitter API error:", e); + } + + return new ArrayList<>(); + }; + } + + @Bean + @ConditionalOnProperty(name = "twitter.friendships.source.type", havingValue = "friends") + public Supplier> friendsSupplier(TwitterFriendshipsSupplierProperties properties, + Twitter twitter, Cursor cursorState) { + return () -> { + try { + PagableResponseList users; + if (properties.getUserId() != null) { + users = twitter.getFriendsList(properties.getUserId(), cursorState.getCursor(), + properties.getCount(), properties.isSkipStatus(), properties.isIncludeUserEntities()); + } + else { // by ScreenName + users = twitter.getFriendsList(properties.getScreenName(), cursorState.getCursor(), + properties.getCount(), properties.isSkipStatus(), properties.isIncludeUserEntities()); + } + + if (users != null) { + cursorState.updateCursor(users.getNextCursor()); + return users; + } + + logger.error(String.format("NULL users response for properties: %s and cursor: %s!", properties, cursorState)); + cursorState.updateCursor(-1); + } + catch (TwitterException e) { + logger.error("Twitter API error:", e); + } + + return new ArrayList<>(); + }; + } + + @Bean + public Function, List> userDeduplicate(MetadataStore metadataStore) { + return users -> { + List uniqueUsers = new ArrayList<>(); + for (User user : users) { + if (metadataStore.get(user.getId() + "") == null) { + metadataStore.put(user.getId() + "", user.getName()); + uniqueUsers.add(user); + } + } + return uniqueUsers; + }; + } + + @Bean + public Supplier> deduplicatedFriendsJsonSupplier(Function, List> userDeduplication, + Supplier> userRetriever, Function> managedJson) { + return () -> userDeduplication.andThen(managedJson).apply(userRetriever.get()); + } +} diff --git a/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/friendships/TwitterFriendshipsSupplierProperties.java b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/friendships/TwitterFriendshipsSupplierProperties.java new file mode 100644 index 00000000..08c2db93 --- /dev/null +++ b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/friendships/TwitterFriendshipsSupplierProperties.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.fn.supplier.twitter.friendships; + +import javax.validation.constraints.AssertTrue; +import javax.validation.constraints.Max; +import javax.validation.constraints.NotNull; +import javax.validation.constraints.Positive; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.validation.annotation.Validated; + + +/** + * @author Christian Tzolov + */ +@ConfigurationProperties("twitter.friendships.source") +@Validated +public class TwitterFriendshipsSupplierProperties { + + public enum FriendshipsRequestType { + /** Friendship query types. */ + followers, friends + } + + /** + * Selects between followers or friends APIs. + */ + @NotNull + private TwitterFriendshipsSupplierProperties.FriendshipsRequestType type = FriendshipsRequestType.followers; + + /** + * The screen name of the user for whom to return results. + */ + private String screenName; + + /** + * The ID of the user for whom to return results. + */ + private Long userId; + + /** + * The number of users to return per page, up to a maximum of 200. Defaults to 20. + */ + @Positive + @Max(200) + private int count = 200; + + /** + * When set to true, statuses will not be included in the returned user objects. + */ + private boolean skipStatus = false; + + /** + * The user object entities node will be disincluded when set to false. + */ + private boolean includeUserEntities = true; + + /** + * API request poll interval in milliseconds. Must be aligned with used APIs rate limits (~ 1 req/ 2 min). + */ + private int pollInterval = 121000; + + public FriendshipsRequestType getType() { + return type; + } + + public void setType(FriendshipsRequestType type) { + this.type = type; + } + + public String getScreenName() { + return screenName; + } + + public void setScreenName(String screenName) { + this.screenName = screenName; + } + + public Long getUserId() { + return userId; + } + + public void setUserId(Long userId) { + this.userId = userId; + } + + public int getCount() { + return count; + } + + public void setCount(int count) { + this.count = count; + } + + public boolean isSkipStatus() { + return skipStatus; + } + + public void setSkipStatus(boolean skipStatus) { + this.skipStatus = skipStatus; + } + + public boolean isIncludeUserEntities() { + return includeUserEntities; + } + + public void setIncludeUserEntities(boolean includeUserEntities) { + this.includeUserEntities = includeUserEntities; + } + + public int getPollInterval() { + return pollInterval; + } + + public void setPollInterval(int pollInterval) { + this.pollInterval = pollInterval; + } + + @AssertTrue(message = "Either userId or screenName must be provided") + public boolean isUserProvided() { + return this.userId != null || this.screenName != null; + } + + @Override + public String toString() { + return "TwitterFriendshipsSourceProperties{" + + "type=" + type + + ", screenName='" + screenName + '\'' + + ", userId=" + userId + + ", count=" + count + + ", skipStatus=" + skipStatus + + ", includeUserEntities=" + includeUserEntities + + ", pollInterval=" + pollInterval + + '}'; + } +} diff --git a/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/message/TwitterMessageSupplierConfiguration.java b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/message/TwitterMessageSupplierConfiguration.java new file mode 100644 index 00000000..cc921f00 --- /dev/null +++ b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/message/TwitterMessageSupplierConfiguration.java @@ -0,0 +1,126 @@ +/* + * 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.message; + +import java.util.ArrayList; +import java.util.List; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import twitter4j.DirectMessage; +import twitter4j.DirectMessageList; +import twitter4j.Twitter; +import twitter4j.TwitterException; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +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.Import; +import org.springframework.integration.metadata.MetadataStore; +import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.messaging.Message; + +/** + * + * @author Christian Tzolov + */ +@EnableConfigurationProperties({ TwitterMessageSupplierProperties.class }) +@Import(TwitterConnectionConfiguration.class) +public class TwitterMessageSupplierConfiguration { + + private static final Log logger = LogFactory.getLog(TwitterMessageSupplierConfiguration.class); + + @Bean + @ConditionalOnMissingBean + public MetadataStore metadataStore() { + return new SimpleMetadataStore(); + } + + @Bean + @ConditionalOnMissingBean + public MessageCursor cursor() { + return new MessageCursor(); + } + + @Bean + public Supplier> directMessagesSupplier(TwitterMessageSupplierProperties properties, + Twitter twitter, MessageCursor cursorState) { + return () -> { + try { + String cs = cursorState.getCursor(); + DirectMessageList messages = (cursorState.getCursor() == null) ? + twitter.getDirectMessages(properties.getCount()) : + twitter.getDirectMessages(properties.getCount(), cursorState.getCursor()); + + if (messages != null) { + cursorState.updateCursor(messages.getNextCursor()); + return messages; + } + + logger.error(String.format("NULL messages response for properties: %s and cursor: %s!", properties, cursorState)); + cursorState.updateCursor(null); + } + catch (TwitterException e) { + logger.error("Twitter API error:", e); + } + + return new ArrayList<>(); + }; + } + + @Bean + public Function, List> messageDeduplicate(MetadataStore metadataStore) { + return messages -> { + List uniqueMessages = new ArrayList<>(); + for (DirectMessage message : messages) { + if (metadataStore.get(message.getId() + "") == null) { + metadataStore.put(message.getId() + "", message.getCreatedAt() + ""); + uniqueMessages.add(message); + } + } + return uniqueMessages; + }; + } + + @Bean + public Supplier> twitterMessageSupplier(Function, List> messageDeduplicate, + Function> managedJson, Supplier> directMessagesSupplier) { + return () -> messageDeduplicate.andThen(managedJson).apply(directMessagesSupplier.get()); + } + + public static class MessageCursor { + private String cursor = null; + + public String getCursor() { + return cursor; + } + + public void updateCursor(String newCursor) { + this.cursor = newCursor; + } + + @Override + public String toString() { + return "Cursor{" + + "cursor=" + cursor + + '}'; + } + } +} diff --git a/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/message/TwitterMessageSupplierProperties.java b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/message/TwitterMessageSupplierProperties.java new file mode 100644 index 00000000..6be5830c --- /dev/null +++ b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/message/TwitterMessageSupplierProperties.java @@ -0,0 +1,45 @@ +/* + * 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.message; + +import javax.validation.constraints.Max; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.validation.annotation.Validated; + + +/** + * @author Christian Tzolov + */ +@ConfigurationProperties("twitter.message.source") +@Validated +public class TwitterMessageSupplierProperties { + + /** + * Max number of events to be returned. 20 default. 50 max. + */ + @Max(50) + private int count = 20; + + public int getCount() { + return count; + } + + public void setCount(int count) { + this.count = count; + } +} diff --git a/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/search/SearchPagination.java b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/search/SearchPagination.java new file mode 100644 index 00000000..c13a15a2 --- /dev/null +++ b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/search/SearchPagination.java @@ -0,0 +1,166 @@ +/* + * 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.search; + +import java.util.List; + +import twitter4j.Status; + +import org.springframework.util.Assert; + +/** + * Searched tweets are ordered from top to bottom by their IDs. The higher the ID, more recent the tweet is. + * + * The search goes backwards - from most recent to the oldest tweets and it uses the sinceId and the maxId to retrieve + * only tweets with IDs in the [sinceId, maxId) range. The -1 stands for unbounded sinceId or maxId. + * + * The first `pageCount` number requests are performed backwards, leaving the bottom boundary (sinceId) unbounded and + * adjusting the upper boundary (maxId) to the lowest tweet ID received. This ensures that no already processed + * tweets are returned. + * + * The pageCounter is used the to count the number of pages retrieved in the pageCount range. Is starts from pageCount + * and goes backward until 0. On 0 the pageCounter is reset back to pageCount. + * + * After performing pageCount number requests (e.g. pageCounter = 0), we start new iteration of searches from the top, + * most recent tweets but now the bottom boundary (sinceId) is adjusted to the max ID received so far. That means that + * only the newly added tweets will be processed + * + * Search pagination with max_id and since_id: https://developer.twitter.com/en/docs/tweets/timelines/guides/working-with-timelines.html + * + * @author Christian Tzolov + */ +public class SearchPagination { + + /** + * Wildcard used as refer as infinity. + */ + public static final long UNBOUNDED = -1; + + /** + * Number of pages to search in history before start form the top again. + * + * Note that the search goes backwards - from most recent to the oldest tweets. + * (eg. maxId == To max ID , sinceId == From min ID) + */ + private final int pageCount; + + /** + * Min tweet ID (e.g. from ID) to be included in the search result + */ + private long sinceId; + + /** + * Keep the max tweet ID in the last pageCount search requests. + */ + private long pageMaxId; + + /** + * Max tweet ID (e.g. To (ID - 1) )to be included in the search result + */ + private long maxId; + + /** + * Current page. Goes backwards from (pageCount-1) to 0. + */ + private int pageCounter; + + /** + * When set it. + */ + boolean searchBackwardsUntilEmptyResponse = false; + + public SearchPagination(int pageCount, boolean searchBackwardsUntilEmptyResponse) { + + Assert.isTrue(pageCount > 0, "At least one page needs to be set but was: " + pageCount); + + this.searchBackwardsUntilEmptyResponse = searchBackwardsUntilEmptyResponse; + this.pageCount = pageCount; + this.sinceId = UNBOUNDED; // == From ID + this.pageMaxId = UNBOUNDED; // == From ID + this.maxId = UNBOUNDED; // == To (ID - 1) + this.pageCounter = pageCount - 1; + } + + public long getSinceId() { + return sinceId; + } + + public long getMaxId() { + return maxId; + } + + public long getPageMaxId() { + return pageMaxId; + } + + public int getPageCounter() { + return pageCounter; + } + + public void update(List tweets) { + + tweets.stream().mapToLong(t -> t.getId()).min() + .ifPresent(tweetsMinId -> { + this.maxId = tweetsMinId - 1; + }); + + tweets.stream().mapToLong(t -> t.getId()).max() + .ifPresent(tweetsMaxId -> { + Assert.isTrue(this.sinceId <= tweetsMaxId, + String.format("MAX_ID (%s) must be bigger then current SINCE_ID(%s)", + tweetsMaxId, this.sinceId)); + this.pageMaxId = Math.max(this.pageMaxId, tweetsMaxId); + }); + + this.countDown(tweets.size()); + } + + private void countDown(int responseSize) { + + if (this.sinceId == UNBOUNDED) { // == first pass before reset + if (this.pageCounter <= 0) { + this.restartSearchFromMostRecent(); + } + } + else { + if (this.searchBackwardsUntilEmptyResponse) { + if (responseSize == 0) { + this.restartSearchFromMostRecent(); + } + } + else if (this.pageCounter <= 0) { + this.restartSearchFromMostRecent(); + } + } + + this.pageCounter--; + } + + private void restartSearchFromMostRecent() { + this.pageCounter = this.pageCount; + + this.sinceId = Math.max(this.sinceId, Math.max(this.maxId + 1, this.pageMaxId)); + + this.maxId = UNBOUNDED; + this.pageMaxId = UNBOUNDED; + } + + public String status() { + return String.format("MaxId: %s, SinceId: %s, Page Counter# %s, pageMaxId: %s", + this.maxId, this.sinceId, this.pageCounter, this.pageMaxId); + } +} diff --git a/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/search/TwitterSearchSupplierConfiguration.java b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/search/TwitterSearchSupplierConfiguration.java new file mode 100644 index 00000000..3910561e --- /dev/null +++ b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/search/TwitterSearchSupplierConfiguration.java @@ -0,0 +1,138 @@ +/* + * 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.search; + +import java.util.List; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import twitter4j.GeoLocation; +import twitter4j.Query; +import twitter4j.QueryResult; +import twitter4j.Status; +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.Import; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * Search pagination with max_id and since_id: https://developer.twitter.com/en/docs/tweets/timelines/guides/working-with-timelines.html . + * + * @author Christian Tzolov + */ + +@EnableConfigurationProperties({ TwitterSearchSupplierProperties.class }) +@Import(TwitterConnectionConfiguration.class) +public class TwitterSearchSupplierConfiguration { + + private static final Log logger = LogFactory.getLog(TwitterSearchSupplierConfiguration.class); + + @Autowired + private TwitterSearchSupplierProperties searchProperties; + + @Autowired + private Twitter twitter; + + @Autowired + private SearchPagination searchPage; + + @Autowired + private Function> json; + + @Bean + public SearchPagination searchPage() { + return new SearchPagination( + this.searchProperties.getPage(), + this.searchProperties.isRestartFromMostRecentOnEmptyResponse()); + } + + + @Bean + public Supplier> twitterSearchSupplier() { + return () -> { + try { + Query query = toQuery(this.searchProperties, this.searchPage); + + QueryResult result = this.twitter.search(query); + + List tweets = result.getTweets(); + + logger.info(String.format("%s, size: %s", this.searchPage.status(), tweets.size())); + + this.searchPage.update(tweets); + + return this.json.apply(tweets); + } + catch (TwitterException e) { + logger.error("Twitter error", e); + } + + return null; + }; + } + + private Query toQuery(TwitterSearchSupplierProperties searchProperties, SearchPagination pagination) { + + Query query = new Query(); + if (searchProperties.getCount() > 0) { + query.count(searchProperties.getCount()); + } + if (StringUtils.hasText(searchProperties.getQuery())) { + query.setQuery(searchProperties.getQuery()); + } + if (StringUtils.hasText(searchProperties.getLang())) { + query.setLang(searchProperties.getLang()); + } + if (StringUtils.hasText(searchProperties.getSince())) { + query.setSince(searchProperties.getSince()); + } + if (searchProperties.getGeocode().isValid()) { + query.setGeoCode( + new GeoLocation( + searchProperties.getGeocode().getLatitude(), + searchProperties.getGeocode().getLongitude()), + searchProperties.getGeocode().getRadius(), + Query.KILOMETERS); + } + + if (searchProperties.getResultType() != Query.ResultType.mixed) { + query.setResultType(searchProperties.getResultType()); + } + + if (pagination.getSinceId() > 0) { + query.setSinceId(pagination.getSinceId()); + } + if (pagination.getMaxId() > 0) { + query.setMaxId(pagination.getMaxId()); + + Assert.isTrue(pagination.getMaxId() >= (pagination.getSinceId() - 1), + String.format("For non empty MAX_ID, The MAX_ID (%s) must always be bigger than [SINCE_ID -1](%s)", + pagination.getMaxId(), (pagination.getSinceId() - 1))); + } + + return query; + } +} diff --git a/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/search/TwitterSearchSupplierProperties.java b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/search/TwitterSearchSupplierProperties.java new file mode 100644 index 00000000..b8b6bfdf --- /dev/null +++ b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/search/TwitterSearchSupplierProperties.java @@ -0,0 +1,198 @@ +/* + * 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.search; + +import javax.validation.constraints.Max; +import javax.validation.constraints.NotEmpty; +import javax.validation.constraints.NotNull; +import javax.validation.constraints.Pattern; +import javax.validation.constraints.Positive; + +import twitter4j.Query; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.validation.annotation.Validated; + +/** + * @author Christian Tzolov + */ +@ConfigurationProperties("twitter.search") +@Validated +public class TwitterSearchSupplierProperties { + + /** + * Search tweets by search query string. + */ + @NotNull + @NotEmpty + private String query; + + /** + * Number of pages (e.g. requests) to search backwards (from most recent to the oldest tweets) before start + * the search from the most recent tweets again. + * The total amount of tweets searched backwards is (page * count) + */ + @Positive + private int page = 3; + + /** + * Number of tweets to return per page (e.g. per single request), up to a max of 100. + */ + @Positive + @Max(100) + private int count = 100; + + /** + * Restricts searched tweets to the given language, given by an ISO 639-1 code. + */ + private String lang = null; + + /** + * If specified, returns tweets with since the given date. Date should be formatted as YYYY-MM-DD. + */ + @Pattern(regexp = "^\\d{4}-\\d{2}-\\d{2}$") + private String since = null; + + /** + * If specified, returns tweets by users located within a given radius (in Km) of the given latitude/longitude, + * where the user's location is taken from their Twitter profile. + * Should be formatted as + */ + private Geocode geocode = new Geocode(); + + /** + * Specifies what type of search results you would prefer to receive. + * The current default is "mixed." Valid values include: + * mixed : Include both popular and real time results in the response. + * recent : return only the most recent results in the response + * popular : return only the most popular results in the response + */ + @NotNull + private Query.ResultType resultType = Query.ResultType.mixed; + + /** + * Restart search from the most recent tweets on empty response. + * Applied only after the first restart (e.g. when since_id != UNBOUNDED) + */ + private boolean restartFromMostRecentOnEmptyResponse = false; + + public String getQuery() { + return query; + } + + public void setQuery(String query) { + this.query = query; + } + + public String getLang() { + return lang; + } + + public void setLang(String lang) { + this.lang = lang; + } + + public int getPage() { + return page; + } + + public void setPage(int page) { + this.page = page; + } + + public int getCount() { + return count; + } + + public void setCount(int count) { + this.count = count; + } + + public String getSince() { + return since; + } + + public void setSince(String since) { + this.since = since; + } + + public Geocode getGeocode() { + return geocode; + } + + public Query.ResultType getResultType() { + return resultType; + } + + public void setResultType(Query.ResultType resultType) { + this.resultType = resultType; + } + + public boolean isRestartFromMostRecentOnEmptyResponse() { + return restartFromMostRecentOnEmptyResponse; + } + + public void setRestartFromMostRecentOnEmptyResponse(boolean restartFromMostRecentOnEmptyResponse) { + this.restartFromMostRecentOnEmptyResponse = restartFromMostRecentOnEmptyResponse; + } + + public static class Geocode { + + /** + * User's latitude. + */ + private double latitude = -1; + + /** + * User's longitude. + */ + private double longitude = -1; + + /** + * Radius (in kilometers) around the (latitude, longitude) point. + */ + private double radius = -1; + + public double getLatitude() { + return latitude; + } + + public void setLatitude(double latitude) { + this.latitude = latitude; + } + + public double getLongitude() { + return longitude; + } + + public void setLongitude(double longitude) { + this.longitude = longitude; + } + + public double getRadius() { + return radius; + } + + public void setRadius(double radius) { + this.radius = radius; + } + + public boolean isValid() { + return this.radius > 0; + } + } +} diff --git a/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierConfiguration.java b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierConfiguration.java new file mode 100644 index 00000000..110bc47c --- /dev/null +++ b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierConfiguration.java @@ -0,0 +1,155 @@ +/* + * 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.util.function.Supplier; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import reactor.core.publisher.Flux; +import twitter4j.StallWarning; +import twitter4j.Status; +import twitter4j.StatusDeletionNotice; +import twitter4j.StatusListener; +import twitter4j.TwitterStream; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.twitter.TwitterConnectionConfiguration; +import org.springframework.cloud.fn.common.twitter.TwitterConnectionProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.integration.channel.FluxMessageChannel; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.MimeTypeUtils; + +/** + * + * @author Christian Tzolov + */ + +@EnableConfigurationProperties({ TwitterStreamSupplierProperties.class, TwitterConnectionProperties.class }) +@Import(TwitterConnectionConfiguration.class) +public class TwitterStreamSupplierConfiguration { + + private static final Log logger = LogFactory.getLog(TwitterStreamSupplierConfiguration.class); + + @Bean + public FluxMessageChannel output() { + return new FluxMessageChannel(); + } + + @Bean + public StatusListener twitterStatusListener(FluxMessageChannel output, TwitterStream twitterStream, + ObjectMapper objectMapper) { + + StatusListener statusListener = new StatusListener() { + + @Override + public void onException(Exception e) { + logger.error("Status Error: ", e); + throw new RuntimeException("Status Error: ", e); + } + + @Override + public void onDeletionNotice(StatusDeletionNotice arg) { + logger.info("StatusDeletionNotice: " + arg); + } + + @Override + public void onScrubGeo(long userId, long upToStatusId) { + logger.info("onScrubGeo: " + userId + ", " + upToStatusId); + } + + @Override + public void onStallWarning(StallWarning warning) { + logger.warn("Stall Warning: " + warning); + throw new RuntimeException("Stall Warning: " + warning); + } + + @Override + public void onStatus(Status status) { + + try { + String json = objectMapper.writeValueAsString(status); + Message message = MessageBuilder.withPayload(json.getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE) + .build(); + output.send(message); + // System.out.println(json); + } + catch (JsonProcessingException e) { + logger.error("Status to JSON conversion error!", e); + throw new RuntimeException("Status to JSON conversion error!", e); + } + } + + @Override + public void onTrackLimitationNotice(int numberOfLimitedStatuses) { + logger.warn("Track Limitation Notice: " + numberOfLimitedStatuses); + } + }; + + twitterStream.addListener(statusListener); + + return statusListener; + } + + @Bean + public Supplier>> twitterStreamSupplier(TwitterStream twitterStream, + FluxMessageChannel output, TwitterStreamSupplierProperties streamProperties) { + + return () -> Flux.from(output) + .doOnSubscribe(subscription -> { + try { + switch (streamProperties.getType()) { + + case filter: + twitterStream.filter(streamProperties.getFilter().toFilterQuery()); + return; + + case sample: + twitterStream.sample(); + return; + + case firehose: + twitterStream.firehose(streamProperties.getFilter().getCount()); + return; + + case link: + twitterStream.links(streamProperties.getFilter().getCount()); + return; + default: + throw new IllegalArgumentException("Unknown stream type:" + streamProperties.getType()); + } + } + catch (Exception e) { + this.logger.error("Filter is not property set"); + } + }) + .doAfterTerminate(() -> { + this.logger.info("Proactive cancel for twitter stream"); + twitterStream.shutdown(); + }) + .doOnError(throwable -> { + this.logger.error(throwable.getMessage(), throwable); + }); + } +} diff --git a/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierProperties.java b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierProperties.java new file mode 100644 index 00000000..7c737641 --- /dev/null +++ b/supplier/twitter-supplier/src/main/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierProperties.java @@ -0,0 +1,267 @@ +/* + * 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.util.ArrayList; +import java.util.List; + +import twitter4j.FilterQuery; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.util.CollectionUtils; +import org.springframework.validation.annotation.Validated; + +/** + * @author Christian Tzolov + */ +@ConfigurationProperties("twitter.stream") +@Validated +public class TwitterStreamSupplierProperties { + + public enum StreamType { + /** Starts listening on random sample of all public statuses. The default access level provides a small + * proportion of the Firehose. */ + sample, + + /** Start consuming public statuses that match one or more filter predicates. At least one predicate parameter, + * follow, locations, or track must be specified. Multiple parameters may be specified which allows most + * clients to use a single connection to the Streaming API. Placing long parameters in the URL may cause the + * request to be rejected for excessive URL length.
+ * The default access level allows up to 200 track keywords, 400 follow userids and 10 1-degree location boxes. + * Increased access levels allow 80,000 follow userids ("shadow" role), 400,000 follow userids ("birddog" role), + * 10,000 track keywords ("restricted track" role), 200,000 track keywords ("partner track" role), + * and 200 10-degree location boxes ("locRestricted" role). Increased track access levels also pass a higher + * proportion of statuses before limiting the stream.*/ + filter, + + /** tarts listening on all public statuses. Available only to approved parties and requires a signed agreement + * to access. */ + firehose, + + /** Starts listening on all public statuses containing links. + * Available only to approved parties and requires a signed agreement to access.*/ + link + } + + private StreamType type = StreamType.sample; + + private Filter filter = new Filter(); + + public Filter getFilter() { + return filter; + } + + public StreamType getType() { + return type; + } + + public void setType(StreamType type) { + this.type = type; + } + + public static class Filter { + + public enum FilterLevel { + /** filter level. */ + all, none, low, medium + } + + /** + * Indicates the number of previous statuses to stream before transitioning to the live stream. + */ + private int count = 0; + + /** + * Specifies the users, by ID, to receive public tweets from. + */ + private List follow; + + /** + * Specifies keywords to track. + */ + private List track; + + + /** + * Locations to track. Internally represented as 2D array. + * Bounding box is invalid: 52.38, 4.90, 51.51, -0.12. The first pair must be the SW corner of the box + */ + private List locations = new ArrayList<>(); + + + /** + * Specifies the tweets language of the stream. + */ + private List language; + + /** + * The filter level limits what tweets appear in the stream to those with a minimum filterLevel attribute value. + * One of either none, low, or medium. + */ + private FilterLevel filterLevel = FilterLevel.all; + + public int getCount() { + return count; + } + + public void setCount(int count) { + this.count = count; + } + + public List getFollow() { + return follow; + } + + public void setFollow(List follow) { + this.follow = follow; + } + + public List getTrack() { + return track; + } + + public void setTrack(List track) { + this.track = track; + } + + public List getLanguage() { + return language; + } + + public void setLanguage(List language) { + this.language = language; + } + + public List getLocations() { + return locations; + } + + public FilterLevel getFilterLevel() { + return filterLevel; + } + + public void setFilterLevel(FilterLevel filterLevel) { + this.filterLevel = filterLevel; + } + + public FilterQuery toFilterQuery() { + + FilterQuery filterQuery = new FilterQuery(); + + filterQuery.count(this.count); + + if (!CollectionUtils.isEmpty(this.track)) { + filterQuery.track(String.join(",", this.track)); + } + + if (!CollectionUtils.isEmpty(this.follow)) { + long[] followIds = new long[this.follow.size()]; + for (int i = 0; i < this.follow.size(); i++) { + followIds[i] = this.follow.get(i); + } + filterQuery.follow(followIds); + } + + if (!CollectionUtils.isEmpty(this.language)) { + filterQuery.language(this.language.toArray(new String[this.language.size()])); + } + + if (!CollectionUtils.isEmpty(this.locations)) { + double[][] bboxLocations = new double[this.locations.size() * 2][2]; + for (int i = 0; i < this.locations.size(); i = i + 2) { + //SW lat, lon + bboxLocations[i][0] = this.locations.get(i).getSw().getLat(); + bboxLocations[i][1] = this.locations.get(i).getSw().getLon(); + //NE lat, lon + bboxLocations[i + 1][0] = this.locations.get(i).getNe().getLat(); + bboxLocations[i + 1][1] = this.locations.get(i).getNe().getLon(); + } + filterQuery.locations(bboxLocations); + } + + if (this.filterLevel != FilterLevel.all) { + filterQuery.filterLevel(this.filterLevel.name()); + } + + return filterQuery; + } + + public boolean isValid() { + return count > 0 || !CollectionUtils.isEmpty(this.track) || !CollectionUtils.isEmpty(this.follow) + || !CollectionUtils.isEmpty(this.language) || this.filterLevel != FilterLevel.all; + } + + public static class BoundingBox { + + /** + * Bounding Box's South-West point (e.g. bottom-left). + */ + private Geocode sw; + + /** + * Bounding Box's North-East point (e.g. top-right). + */ + private Geocode ne; + + public Geocode getSw() { + return sw; + } + + public void setSw(Geocode sw) { + this.sw = sw; + } + + public Geocode getNe() { + return ne; + } + + public void setNe(Geocode ne) { + this.ne = ne; + } + } + + public static class Geocode { + + /** + * latitude. + */ + private double lat = -1; + + /** + * longitude. + */ + private double lon = -1; + + + public double getLat() { + return lat; + } + + public void setLat(double lat) { + this.lat = lat; + } + + public double getLon() { + return lon; + } + + public void setLon(double lon) { + this.lon = lon; + } + } + } +} diff --git a/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/search/SearchPaginationTests.java b/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/search/SearchPaginationTests.java new file mode 100644 index 00000000..67e48ce9 --- /dev/null +++ b/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/search/SearchPaginationTests.java @@ -0,0 +1,349 @@ +/* + * 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.search; + +import java.util.ArrayList; +import java.util.Date; +import java.util.List; + +import org.junit.Test; +import twitter4j.GeoLocation; +import twitter4j.HashtagEntity; +import twitter4j.MediaEntity; +import twitter4j.Place; +import twitter4j.RateLimitStatus; +import twitter4j.Scopes; +import twitter4j.Status; +import twitter4j.SymbolEntity; +import twitter4j.URLEntity; +import twitter4j.User; +import twitter4j.UserMentionEntity; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.cloud.fn.supplier.twitter.status.search.SearchPagination.UNBOUNDED; + +/** + * @author Christian Tzolov + */ +public class SearchPaginationTests { + + @Test + public void tests1() { + + int maxCountPerRequest = 5; + + int count = 10; + + int pageCount = count / maxCountPerRequest; + + assertThat(pageCount).isEqualTo(2); + + SearchPagination pagination = new SearchPagination(pageCount, false); + + assertThat(pagination.getSinceId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 1); + + pagination.update(tweets(10, 8, 1)); + + assertThat(pagination.getSinceId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getMaxId()).isEqualTo(8L - 1); // = min - 1 + assertThat(pagination.getPageMaxId()).isEqualTo(10L); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 2); + + pagination.update(tweets(7, 4, 1)); + + assertThat(pagination.getSinceId()).isEqualTo(10L); + assertThat(pagination.getMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 1); + + pagination.update(tweets(20, 14, 1)); + + assertThat(pagination.getSinceId()).isEqualTo(10L); + assertThat(pagination.getMaxId()).isEqualTo(14L - 1); + assertThat(pagination.getPageMaxId()).isEqualTo(20L); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 2); + + pagination.update(tweets(13, 8, 1)); + + assertThat(pagination.getSinceId()).isEqualTo(20L); + assertThat(pagination.getMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 1); + } + + @Test + public void tests2() { + + int maxCountPerRequest = 5; + + int count = 10; + + int pageCount = count / maxCountPerRequest; + + assertThat(pageCount).isEqualTo(2); + + SearchPagination pagination = new SearchPagination(pageCount, true); + + assertThat(pagination.getSinceId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 1); + + pagination.update(tweets(10, 8, 1)); + + assertThat(pagination.getSinceId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getMaxId()).isEqualTo(8L - 1); // = min - 1 + assertThat(pagination.getPageMaxId()).isEqualTo(10L); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 2); + + pagination.update(tweets(7, 4, 1)); + + // Restart from Most Recent due to pageCounter == 0 while no reset have been performed so ar + assertThat(pagination.getSinceId()).isEqualTo(10L); + assertThat(pagination.getMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 1); + + pagination.update(tweets(20, 14, 1)); + + assertThat(pagination.getSinceId()).isEqualTo(10L); + assertThat(pagination.getMaxId()).isEqualTo(14L - 1); + assertThat(pagination.getPageMaxId()).isEqualTo(20L); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 2); + + pagination.update(tweets(13, 8, 1)); + + assertThat(pagination.getSinceId()).isEqualTo(10L); + assertThat(pagination.getMaxId()).isEqualTo(8L - 1); + assertThat(pagination.getPageMaxId()).isEqualTo(20L); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 3); + + pagination.update(new ArrayList<>()); // EMPTY + + // Restart from Most Recent on Empty Response + assertThat(pagination.getSinceId()).isEqualTo(20L); + assertThat(pagination.getMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageMaxId()).isEqualTo(UNBOUNDED); + assertThat(pagination.getPageCounter()).isEqualTo(pageCount - 1); + } + + public List tweets(int to, int from, int step) { + List tweets = new ArrayList<>(to - from); + for (int i = to; i >= from; i = i - step) { + tweets.add(new MyStatus(i)); + } + + return tweets; + } + + public static class MyStatus implements Status { + + private long id; + + public MyStatus(long id) { + this.id = id; + } + + @Override + public Date getCreatedAt() { + return null; + } + + @Override + public long getId() { + return this.id; + } + + @Override + public String getText() { + return null; + } + + @Override + public int getDisplayTextRangeStart() { + return 0; + } + + @Override + public int getDisplayTextRangeEnd() { + return 0; + } + + @Override + public String getSource() { + return null; + } + + @Override + public boolean isTruncated() { + return false; + } + + @Override + public long getInReplyToStatusId() { + return 0; + } + + @Override + public long getInReplyToUserId() { + return 0; + } + + @Override + public String getInReplyToScreenName() { + return null; + } + + @Override + public GeoLocation getGeoLocation() { + return null; + } + + @Override + public Place getPlace() { + return null; + } + + @Override + public boolean isFavorited() { + return false; + } + + @Override + public boolean isRetweeted() { + return false; + } + + @Override + public int getFavoriteCount() { + return 0; + } + + @Override + public User getUser() { + return null; + } + + @Override + public boolean isRetweet() { + return false; + } + + @Override + public Status getRetweetedStatus() { + return null; + } + + @Override + public long[] getContributors() { + return new long[0]; + } + + @Override + public int getRetweetCount() { + return 0; + } + + @Override + public boolean isRetweetedByMe() { + return false; + } + + @Override + public long getCurrentUserRetweetId() { + return 0; + } + + @Override + public boolean isPossiblySensitive() { + return false; + } + + @Override + public String getLang() { + return null; + } + + @Override + public Scopes getScopes() { + return null; + } + + @Override + public String[] getWithheldInCountries() { + return new String[0]; + } + + @Override + public long getQuotedStatusId() { + return 0; + } + + @Override + public Status getQuotedStatus() { + return null; + } + + @Override + public URLEntity getQuotedStatusPermalink() { + return null; + } + + @Override + public int compareTo(Status o) { + return 0; + } + + @Override + public UserMentionEntity[] getUserMentionEntities() { + return new UserMentionEntity[0]; + } + + @Override + public URLEntity[] getURLEntities() { + return new URLEntity[0]; + } + + @Override + public HashtagEntity[] getHashtagEntities() { + return new HashtagEntity[0]; + } + + @Override + public MediaEntity[] getMediaEntities() { + return new MediaEntity[0]; + } + + @Override + public SymbolEntity[] getSymbolEntities() { + return new SymbolEntity[0]; + } + + @Override + public RateLimitStatus getRateLimitStatus() { + return null; + } + + @Override + public int getAccessLevel() { + return 0; + } + } +} diff --git a/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierTests.java b/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierTests.java new file mode 100644 index 00000000..809b5b35 --- /dev/null +++ b/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterStreamSupplierTests.java @@ -0,0 +1,211 @@ +/* + * 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.util.concurrent.TimeUnit; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockserver.client.MockServerClient; +import org.mockserver.integration.ClientAndServer; +import org.mockserver.model.Header; +import org.mockserver.model.HttpRequest; +import org.mockserver.model.StringBody; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; +import twitter4j.conf.ConfigurationBuilder; + +import org.springframework.beans.factory.annotation.Autowired; +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.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.Primary; +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 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 + */ +@RunWith(SpringRunner.class) +@SpringBootTest( + webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = { + "twitter.connection.consumerKey=consumerKey666", + "twitter.connection.consumerSecret=consumerSecret666", + "twitter.connection.accessToken=accessToken666", + "twitter.connection.accessTokenSecret=accessTokenSecret666" + }) +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) +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 ClientAndServer mockServer; + + private static MockServerClient mockClient; + private static HttpRequest streamFilterRequest; + private static HttpRequest streamSampleRequest; + private static HttpRequest streamFirehoseRequest; + + @Autowired + protected Supplier>> twitterStreamSupplier; + + @BeforeClass + public static void startServer() { + + mockServer = ClientAndServer.startClientAndServer(MOCK_SERVER_PORT); + + mockClient = new MockServerClient(MOCK_SERVER_IP, MOCK_SERVER_PORT); + + streamFilterRequest = mockClientRecordRequest(request() + .withMethod("POST") + .withPath("/stream/statuses/filter.json") + .withBody(new StringBody("count=0&track=Java%2CPython&stall_warnings=true"))); + + streamSampleRequest = mockClientRecordRequest(request() + .withMethod("GET") + .withPath("/stream/statuses/sample.json")); + + streamFirehoseRequest = mockClientRecordRequest(request() + .withMethod("POST") + .withPath("/stream/statuses/links.json") + .withBody(new StringBody("count=0&stall_warnings=true"))); + + streamFirehoseRequest = mockClientRecordRequest(request() + .withMethod("POST") + .withPath("/stream/statuses/firehose.json") + .withBody(new StringBody("count=0&stall_warnings=true"))); + } + + @AfterClass + public static void stopServer() { + mockServer.stop(); + } + + private static HttpRequest mockClientRecordRequest(HttpRequest request) { + mockClient.when(request, /*unlimited())*/ 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/stream_test_1.json")) + .withDelay(TimeUnit.SECONDS, 10)); + return request; + } + + @TestPropertySource(properties = { + "twitter.stream.type=sample" + }) + public static class TwitterStreamSampleTests extends TwitterStreamSupplierTests { + + @Test + public void testOne() { + final Flux> messageFlux = twitterStreamSupplier.get(); + + final StepVerifier stepVerifier = StepVerifier.create(messageFlux) + .assertNext((message) -> assertThat(new String((byte[]) message.getPayload())) + .contains("\"id\":1075751718749659136")) + .thenCancel() + .verifyLater(); + + stepVerifier.verify(); + + mockClient.verify(streamSampleRequest, once()); + } + } + + @TestPropertySource(properties = { + "twitter.stream.type=filter", + "twitter.stream.filter.track=Java,Python" + }) + public static class TwitterStreamFilterTests extends TwitterStreamSupplierTests { + + @Test + public void testOne() { + final Flux> messageFlux = twitterStreamSupplier.get(); + + final StepVerifier stepVerifier = StepVerifier.create(messageFlux) + .assertNext((message) -> assertThat(new String((byte[]) message.getPayload())) + .contains("\"id\":1075751718749659136")) + .thenCancel() + .verifyLater(); + + stepVerifier.verify(); + + mockClient.verify(streamFilterRequest, once()); + } + } + + @TestPropertySource(properties = { + "twitter.stream.type=firehose" + }) + public static class TwitterStreamFirehoseTests extends TwitterStreamSupplierTests { + + @Test + public void testOne() { + final Flux> messageFlux = twitterStreamSupplier.get(); + + final StepVerifier stepVerifier = StepVerifier.create(messageFlux) + .assertNext((message) -> assertThat(new String((byte[]) message.getPayload())) + .contains("\"id\":1075751718749659136")) + .thenCancel() + .verifyLater(); + + stepVerifier.verify(); + + mockClient.verify(streamFirehoseRequest, once()); + } + } + + @SpringBootConfiguration + @EnableAutoConfiguration + @Import(TwitterStreamSupplierConfiguration.class) + public static class TestTwitterStreamSourceApplication { + + @Bean + @Primary + public twitter4j.conf.Configuration twitterConfiguration2(TwitterConnectionProperties properties, + Function toConfigurationBuilder) { + + Function mockedConfiguration = + toConfigurationBuilder.andThen( + new TwitterTestUtils().mockTwitterUrls( + String.format("http://%s:%s", MOCK_SERVER_IP, MOCK_SERVER_PORT))); + + return mockedConfiguration.apply(properties).build(); + } + } +} diff --git a/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterTestUtils.java b/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/TwitterTestUtils.java new file mode 100644 index 00000000..69305887 --- /dev/null +++ b/supplier/twitter-supplier/src/test/java/org/springframework/cloud/fn/supplier/twitter/status/stream/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.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); + } + } + +} diff --git a/supplier/twitter-supplier/src/test/resources/response/stream_test_1.json b/supplier/twitter-supplier/src/test/resources/response/stream_test_1.json new file mode 100644 index 00000000..bb831206 --- /dev/null +++ b/supplier/twitter-supplier/src/test/resources/response/stream_test_1.json @@ -0,0 +1,2 @@ +{"created_at":"Thu Dec 20 13:56:10 +0000 2018","id":1075751718749659136,"id_str":"1075751718749659136","text":"Codementor: Kubernetes for Python Developers: Part 1\n#100daysofcode #python https:\/\/t.co\/VviTgpFpge","source":"\u003ca href=\"https:\/\/ifttt.com\" rel=\"nofollow\"\u003eIFTTT\u003c\/a\u003e","truncated":false,"in_reply_to_status_id":null,"in_reply_to_status_id_str":null,"in_reply_to_user_id":null,"in_reply_to_user_id_str":null,"in_reply_to_screen_name":null,"user":{"id":859252650512072704,"id_str":"859252650512072704","name":"Freelancing|WebDev","screen_name":"FreelanceForBTC","location":"THE NET","url":"http:\/\/bit.ly\/BTCFREELANCING","description":"This twitter is designed to give the best information from web development to Blockchain.\n|YouTube's |Jobs |learning contented","translator_type":"none","protected":false,"verified":false,"followers_count":146,"friends_count":118,"listed_count":4,"favourites_count":445,"statuses_count":7528,"created_at":"Tue May 02 03:46:10 +0000 2017","utc_offset":null,"time_zone":null,"geo_enabled":false,"lang":"en","contributors_enabled":false,"is_translator":false,"profile_background_color":"F5F8FA","profile_background_image_url":"","profile_background_image_url_https":"","profile_background_tile":false,"profile_link_color":"1DA1F2","profile_sidebar_border_color":"C0DEED","profile_sidebar_fill_color":"DDEEF6","profile_text_color":"333333","profile_use_background_image":true,"profile_image_url":"http:\/\/pbs.twimg.com\/profile_images\/1071607798771793920\/UegrCs84_normal.jpg","profile_image_url_https":"https:\/\/pbs.twimg.com\/profile_images\/1071607798771793920\/UegrCs84_normal.jpg","profile_banner_url":"https:\/\/pbs.twimg.com\/profile_banners\/859252650512072704\/1544326203","default_profile":true,"default_profile_image":false,"following":null,"follow_request_sent":null,"notifications":null},"geo":null,"coordinates":null,"place":null,"contributors":null,"is_quote_status":false,"quote_count":0,"reply_count":0,"retweet_count":0,"favorite_count":0,"entities":{"hashtags":[{"text":"100daysofcode","indices":[53,67]},{"text":"python","indices":[68,75]}],"urls":[{"url":"https:\/\/t.co\/VviTgpFpge","expanded_url":"https:\/\/ift.tt\/2GxdMZF","display_url":"ift.tt\/2GxdMZF","indices":[76,99]}],"user_mentions":[],"symbols":[]},"favorited":false,"retweeted":false,"possibly_sensitive":false,"filter_level":"low","lang":"ca","timestamp_ms":"1545314170907"} +