Twitter functional apps

* Streaming source * Add twitter search and messasge sources * Add more twitter source, sink and processors * Add IT tests for twitter update sink * Add IT tests for twitter message sink * Add IT tests for twitter trend processor * twitter suppliers readme * twitter consumers readme * twitter functions readme * Disable TwitterStreamSourceTests
2020-06-18 16:09:56 +02:00
parent dab3da942d
commit 1892736b98
34 changed files with 3852 additions and 0 deletions
--- a/common/twitter-common/pom.xml
+++ b/common/twitter-common/pom.xml
@@ -0,0 +1,56 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>twitter-common</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>twitter-common</name>
+    <description>Twitter common</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <properties>
+        <twitter4j.version>4.0.7</twitter4j.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.twitter4j</groupId>
+            <artifactId>twitter4j-stream</artifactId>
+            <version>${twitter4j.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-json</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-ip</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>
--- 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
@@ -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 + '}';
+	}
+}
--- 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
@@ -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 {
+	}
+}
--- 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
@@ -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<TwitterConnectionProperties, ConfigurationBuilder> 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<TwitterConnectionProperties, ConfigurationBuilder> 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<Object, Message<byte[]>> 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<Object, Object> rawJsonExtractor() {
+		return response -> {
+			if (response instanceof List) {
+				List responses = (List) response;
+				List<String> rawJsonList = new ArrayList<>();
+				for (Object object : responses) {
+					rawJsonList.add(TwitterObjectFactory.getRawJSON(object));
+				}
+				return rawJsonList;
+			}
+			else {
+				return TwitterObjectFactory.getRawJSON(response);
+			}
+		};
+	}
+
+	@Bean
+	public Function<Object, Message<byte[]>> managedJson(TwitterConnectionProperties properties,
+			Function<Object, Object> rawJsonExtractor, Function<Object, Message<byte[]>> json) {
+		return list -> (properties.isRawJson()) ? rawJsonExtractor.andThen(json).apply(list) : json.apply(list);
+	}
+}
--- 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
@@ -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;
+	}
+}
--- a/consumer/twitter-consumer/README.adoc
+++ 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<StatusUpdate> updateStatus` - if you have an `StatusUpdate` instance you can use the `updateStatus` to apply it.
+
+- `Function<Message<?>, StatusUpdate> toStatusUpdateQuery` - function that converts a `Message<?>`  text into a `StatusUpdate` instance using the `TwitterUpdateConsumerProperties` properties.
+
+- `Consumer<Message<?>> 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<Message<?>> 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<Message<?>> 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].
+
--- a/consumer/twitter-consumer/pom.xml
+++ b/consumer/twitter-consumer/pom.xml
@@ -0,0 +1,57 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>twitter-consumer</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>twitter-consumer</name>
+    <description>twitter consumer</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>twitter-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>payload-converter-function</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>
--- 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
@@ -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<Message<?>> 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);
+			}
+		};
+	}
+}
--- 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
@@ -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;
+		}
+	}
+}
--- 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
@@ -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<Message<?>> 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);
+			}
+		};
+	}
+}
--- 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
@@ -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;
+	}
+}
--- 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
@@ -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<StatusUpdate> 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<Message<?>, 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<Message<?>> twitterStatusUpdateConsumer(Function<Message<?>, StatusUpdate> statusUpdateQuery,
+			Consumer<StatusUpdate> updateStatus) {
+		return message -> updateStatus.accept(statusUpdateQuery.apply(message));
+	}
+}
--- 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
@@ -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;
+		}
+	}
+}
--- 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
@@ -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<StatusUpdate> 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<Message<?>, 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);
+	}
+}
--- a/function/twitter-function/README.adoc
+++ 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<?>, Message<byte[]>> 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
+
--- a/function/twitter-function/pom.xml
+++ b/function/twitter-function/pom.xml
@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<artifactId>twitter-function</artifactId>
+	<version>1.0.0-SNAPSHOT</version>
+	<name>twitter-function</name>
+	<description>twitter functions</description>
+
+	<parent>
+		<groupId>org.springframework.cloud.fn</groupId>
+		<artifactId>spring-functions-parent</artifactId>
+		<version>1.0.0-SNAPSHOT</version>
+		<relativePath>../../spring-functions-parent</relativePath>
+	</parent>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud.fn</groupId>
+			<artifactId>twitter-common</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-integration</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.junit.vintage</groupId>
+					<artifactId>junit-vintage-engine</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-configuration-processor</artifactId>
+			<scope>provided</scope>
+		</dependency>
+	</dependencies>
+
+</project>
--- 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
@@ -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<Message<?>, 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<Message<?>, List<Location>> 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<?>, Message<byte[]>> trendOrTrendLocationsFunction(
+			Function<Object, Message<byte[]>> managedJson, Function<Message<?>, Trends> trend,
+			TwitterTrendFunctionProperties properties, Function<Message<?>,
+			List<Location>> closestOrAvailableTrends) {
+
+		return (properties.getTrendQueryType() == TwitterTrendFunctionProperties.TrendQueryType.trend) ?
+				trend.andThen(managedJson) : closestOrAvailableTrends.andThen(managedJson);
+	}
+}
--- 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
@@ -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;
+		}
+	}
+}
--- a/pom.xml
+++ b/pom.xml
@@ -50,6 +50,7 @@
 		<module>common/metadata-store-common</module>
 		<module>common/mqtt-common</module>
 		<module>common/tcp-common</module>
+		<module>common/twitter-common</module>


 		<module>consumer/cassandra-consumer</module>
@@ -67,6 +68,7 @@
 		<module>consumer/tcp-consumer</module>
 		<module>consumer/websocket-consumer</module>
 		<module>consumer/s3-consumer</module>
+		<module>consumer/twitter-consumer</module>

 		<module>function/filter-function</module>
 		<module>function/header-enricher-function</module>
@@ -76,6 +78,7 @@
 		<module>function/splitter-function</module>
 		<module>function/tasklauncher-function</module>
 		<module>function/task-launch-request-function</module>
+		<module>function/twitter-function</module>

 		<module>supplier/file-supplier</module>
 		<module>supplier/ftp-supplier</module>
@@ -90,6 +93,7 @@
 		<module>supplier/rabbit-supplier</module>
 		<module>supplier/websocket-supplier</module>
 		<module>supplier/s3-supplier</module>
+		<module>supplier/twitter-supplier</module>

 		<module>spring-functions-parent</module>
 	</modules>
--- a/supplier/twitter-supplier/README.adoc
+++ 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<Message<byte[]>>`.
+
+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<Flux<Message<?>>>`.
+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<Flux<Message<?>>>`.
+
+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<Message<byte[]>>`.
+
+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<List<User>>`.
+
+- `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<Message<byte[]>>`.
+
+
+### 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]
--- a/supplier/twitter-supplier/pom.xml
+++ b/supplier/twitter-supplier/pom.xml
@@ -0,0 +1,80 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>twitter-supplier</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>twitter-supplier</name>
+    <description>twitter suppliers</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>twitter-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-jms</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>javax.jms</groupId>
+            <artifactId>javax.jms-api</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.activemq</groupId>
+            <artifactId>activemq-broker</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mock-server</groupId>
+            <artifactId>mockserver-netty</artifactId>
+            <version>5.10</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mock-server</groupId>
+            <artifactId>mockserver-client-java</artifactId>
+            <version>5.10</version>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+</project>
--- 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
@@ -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<List<User>> followersSupplier(TwitterFriendshipsSupplierProperties properties,
+			Twitter twitter, Cursor cursorState) {
+		return () -> {
+			try {
+				PagableResponseList<User> 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<List<User>> friendsSupplier(TwitterFriendshipsSupplierProperties properties,
+			Twitter twitter, Cursor cursorState) {
+		return () -> {
+			try {
+				PagableResponseList<User> 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<User>, List<User>> userDeduplicate(MetadataStore metadataStore) {
+		return users -> {
+			List<User> 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<Message<byte[]>> deduplicatedFriendsJsonSupplier(Function<List<User>, List<User>> userDeduplication,
+			Supplier<List<User>> userRetriever, Function<Object, Message<byte[]>> managedJson) {
+		return () -> userDeduplication.andThen(managedJson).apply(userRetriever.get());
+	}
+}
--- 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
@@ -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 +
+				'}';
+	}
+}
--- 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
@@ -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<List<DirectMessage>> 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<DirectMessage>, List<DirectMessage>> messageDeduplicate(MetadataStore metadataStore) {
+		return messages -> {
+			List<DirectMessage> 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<Message<byte[]>> twitterMessageSupplier(Function<List<DirectMessage>, List<DirectMessage>> messageDeduplicate,
+			Function<Object, Message<byte[]>> managedJson, Supplier<List<DirectMessage>> 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 +
+					'}';
+		}
+	}
+}
--- 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
@@ -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;
+	}
+}
--- 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
@@ -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<Status> 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);
+	}
+}
--- 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
@@ -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<Object, Message<byte[]>> json;
+
+	@Bean
+	public SearchPagination searchPage() {
+		return new SearchPagination(
+				this.searchProperties.getPage(),
+				this.searchProperties.isRestartFromMostRecentOnEmptyResponse());
+	}
+
+
+	@Bean
+	public Supplier<Message<byte[]>> twitterSearchSupplier() {
+		return () -> {
+			try {
+				Query query = toQuery(this.searchProperties, this.searchPage);
+
+				QueryResult result = this.twitter.search(query);
+
+				List<Status> 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;
+	}
+}
--- 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
@@ -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 <a href="http://en.wikipedia.org/wiki/ISO_639-1">ISO 639-1 code</a>.
+	 */
+	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;
+		}
+	}
+}
--- 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
@@ -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<byte[]> 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<Flux<Message<?>>> 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);
+				});
+	}
+}
--- 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
@@ -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.<br>
+		 * 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<Long> follow;
+
+		/**
+		 * Specifies keywords to track.
+		 */
+		private List<String> 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<BoundingBox> locations = new ArrayList<>();
+
+
+		/**
+		 * Specifies the tweets language of the stream.
+		 */
+		private List<String> 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<Long> getFollow() {
+			return follow;
+		}
+
+		public void setFollow(List<Long> follow) {
+			this.follow = follow;
+		}
+
+		public List<String> getTrack() {
+			return track;
+		}
+
+		public void setTrack(List<String> track) {
+			this.track = track;
+		}
+
+		public List<String> getLanguage() {
+			return language;
+		}
+
+		public void setLanguage(List<String> language) {
+			this.language = language;
+		}
+
+		public List<BoundingBox> 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;
+			}
+		}
+	}
+}
--- 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
@@ -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<Status> tweets(int to, int from, int step) {
+		List<Status> 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;
+		}
+	}
+}
--- 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
@@ -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<Flux<Message<?>>> 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<Message<?>> 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<Message<?>> 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<Message<?>> 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<TwitterConnectionProperties, ConfigurationBuilder> toConfigurationBuilder) {
+
+			Function<TwitterConnectionProperties, ConfigurationBuilder> mockedConfiguration =
+					toConfigurationBuilder.andThen(
+							new TwitterTestUtils().mockTwitterUrls(
+									String.format("http://%s:%s", MOCK_SERVER_IP, MOCK_SERVER_PORT)));
+
+			return mockedConfiguration.apply(properties).build();
+		}
+	}
+}
--- 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
@@ -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<ConfigurationBuilder, ConfigurationBuilder> mockTwitterUrls(String baseUrl) {
+		return configBuilder -> {
+			configBuilder.setRestBaseURL(baseUrl + "/");
+			configBuilder.setStreamBaseURL(baseUrl + "/stream/");
+			configBuilder.setUserStreamBaseURL(baseUrl + "/user/");
+			configBuilder.setSiteStreamBaseURL(baseUrl + "/site/");
+			configBuilder.setUploadBaseURL(baseUrl + "/upload/");
+
+			configBuilder.setOAuthAccessTokenURL(baseUrl + "/oauth/access_token");
+			configBuilder.setOAuthAuthenticationURL(baseUrl + "/oauth/authenticate");
+			configBuilder.setOAuthAuthorizationURL(baseUrl + "/oauth/authorize");
+			configBuilder.setOAuthRequestTokenURL(baseUrl + "/oauth/request_token");
+			configBuilder.setOAuth2TokenURL(baseUrl + "/oauth2/token");
+			configBuilder.setOAuth2InvalidateTokenURL(baseUrl + "/oauth2/invalidate_token");
+
+			return configBuilder;
+		};
+	}
+
+	/**
+	 * Load Spring Resource as String.
+	 * @param resourcePath Resource path (accepts file:// , classpath:// and http:// uri schemas)
+	 * @return Returns text (UTF8) representation of the resource pointed by the resourcePath
+	 */
+	public static String asString(String resourcePath) {
+		try {
+			return StreamUtils.copyToString(new DefaultResourceLoader().getResource(resourcePath).getInputStream(),
+					Charset.forName("UTF-8"));
+		}
+		catch (IOException e) {
+			throw new RuntimeException("Can not load resource:" + resourcePath, e);
+		}
+	}
+
+}
--- a/supplier/twitter-supplier/src/test/resources/response/stream_test_1.json
+++ 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"}
+
				`@@ -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"}