Spring Integration<\/a> has been known to have a myriad of connectors for interacting with external systems. Twitter was no exception and for a long time, as Spring Social<\/a> had been an out-of-the-box solution which Spring Integration leveraged in order to connect to social networks.<\/p>\n Unfortunately, Spring Social has reached its end of life<\/a>, the project is now in maintenance mode.<\/strong> The reason why the Spring Team decided that they wouldn\u2019t develop Spring Social further was that it became tedious to keep the API bindings in sync with the APIs of social networks.<\/p>\n Other than that, after Spring Framework 5 had been released, developers wished to leverage its reactive programming model and that would have required the team to re-implement a reactive Spring Social bindings next to the existing one.<\/p>\n Developers are now advised to either implement their own binding or use one of the purpose built libraries to connect to social networks.<\/strong><\/p>\n The fact the Spring Social is now in maintenance mode forced the Spring Integration team to move the Twitter support module from the main project to the extensions. As Spring Social isn\u2019t going to receive updates, it\u2019s going to be built upon an earlier Spring Framework version. That would lead to class path conflict and would also hamper the development of Spring Integration.<\/p>\n Therefore, as of Spring Integration 5.1, the Twitter module is available as an extension<\/a>.<\/strong><\/p>\n Twitter4J<\/strong><\/a> is an unofficial Java libary for Twitter\u2019s API developed and maintained by Yusuke Yamamoto<\/a>. The official HBC<\/a> library (built by Twitter)<\/strong> is a Java HTTP Client for consuming Twitter\u2019s Streaming API. The latter hasn\u2019t seen major updates since 2016, while Twitter4J is receiving regular updates.<\/p>\n Implementing your own API binding is also an option.<\/strong> In Spring based projects using RestTemplate<\/a> and is definitely an option and it\u2019s an easy way to make REST calls.<\/p>\n This guide uses Twitter4J in streaming mode in a way that can be integrated into a Spring Integration message flow.<\/strong><\/p>\n In a nutshell, your app opens a single connection to to Twitter\u2019s API and new results are sent through that connection whenever new matches occur<\/strong>. In contrast, the other way around is delivering data in batches through repeated requests to a REST API.<\/p>\n Streaming provides a low-latency delivery mechanism<\/strong> that can support very high throughput without having to deal with rate limiting.<\/p>\n The example project, which demonstrates the integration of Twitter\u2019s Streaming API into a Spring Integration message flow, is available on GitHub<\/strong>: https:\/\/github.com\/springuni\/springuni-examples\/tree\/master\/spring-integration\/twitter-streaming<\/a>.<\/p>\n As Spring Social is EOL now, we won\u2019t build upon it. All we pull in are spring-integration-core<\/a> and twitter4j-stream<\/a>.<\/p>\n This project also uses Lombok<\/a> and Spring Boot testing<\/a> support, but these are optional. Spring Integration provides support for implementing inbound message components<\/strong>. They\u2019re divided into polling and listening behaviors<\/strong>.<\/p>\n The original Inbound Twitter Channel Adapter<\/strong>, that one which relies builds upon Spring Social and is now moved to the extensions, is a polling consumer<\/strong>. That is, you have to provide a poller configuration to use it. On the other hand, Twitter enforces Rate Limits in order to manage how often application can fetch updates. You should have taken Rate Limiting into consideration when the old Twitter Channel adapter was used, so that your configured poller intervals have been in compliance with the Twitter policies.<\/p>\n On the other hand, the listening inbound components are simpler and typically require only MessageProducerSupport<\/a> to be implemented.<\/strong> Such a listening component looks like this.<\/p>\n There are only two required elements:<\/strong><\/p>\n Optionally you might want to take control over the component\u2019s initialization and manage its lifecycle.<\/p>\n As Twitter\u2019s Streaming API is inherently message-driven, the listening behavior is a natural fit.<\/strong> Let\u2019s see how Twitter4J can be incorporated in such a context.<\/p>\n Twitter4J manages the nuances of connection handing and receiving updates from Twitter\u2019s Streaming API. All we need to do is acquire a Streaming examples on Twitter4J\u2019s website suggest that a Spring\u2019s Although we could also expose it as a regular bean without being created by a That\u2019s going to be the responsibility of our custom Lifecycle methods provided by Although Spring could auto-wire components, I still prefer controlling dependencies with manual configuration.<\/p>\n Important part here is that how our custom message producer integrates with a message flow. Basically, we don\u2019t need to do anything, other than listing to messages at the producer\u2019s output channel.<\/p>\n Only Chuck Norris tests code in production. However, ordinary mortal folks like you and me, we do write test cases.<\/p>\n I like Twitter4J\u2019s design, because it leverages interfaces.<\/strong> Most of the important parts of the library are exposed as ordinary interfaces. 1.1 Spring Social EOL<\/h3>\n
1.2 Spring Integration\u2019s Twitter module moved to extensions<\/h3>\n
1.3 What are the alternatives?<\/h3>\n
1.4 How does Twitter Streaming work?<\/h3>\n
2. Example project<\/h2>\n
2.1. Maven Dependencies<\/h3>\n
<dependencies>\n <dependency>\n <groupId>org.springframework.integration<\/groupId>\n <artifactId>spring-integration-core<\/artifactId>\n <\/dependency>\n\n <dependency>\n <groupId>org.twitter4j<\/groupId>\n <artifactId>twitter4j-stream<\/artifactId>\n <version>4.0.1<\/version>\n <\/dependency>\n <\/dependencies><\/pre>\n
2.3. Listenable message source with Spring Integration<\/h3>\n
public class MyMessageProducer extends MessageProducerSupport {\n\n public MyMessageProducer(MessageChannel outputChannel) {\n \/\/ Defining an output channel is required\n setOutputChannel(outputChannel);\n }\n\n @Override\n protected void onInit() {\n super.onInit();\n \/\/ Custom initialization - if applicable - comes here\n }\n\n @Override\n public void doStart() {\n \/\/ Lifecycle method for starting receiving messages\n }\n\n @Override\n public void doStop() {\n \/\/ Lifecycle method for stopping receiving messages\n }\n\n private void receiveMessage() {\n \/\/ Receive data from upstream service\n SomeData data = ...;\n\n \/\/ Convert it to a message as appropriate and send it out\n this.sendMessage(MessageBuilder.withPayload(data).build());\n }\n\n}<\/pre>\n\n
sendMessage<\/code> has to be called whenever the component receives a message<\/li>\n<\/ul>\n2.4. Connect to Twitter Streaming API with Twitter4J<\/h3>\n
TwitterStream<\/code> instance, attach a listener and define filtering.<\/p>\n2.4.1. Instantiate
TwitterStream<\/code><\/h2>\nTwitterStream<\/code><\/a> instance should be created through TwitterStreamFactory<\/code><\/a>. That makes perfectly sense, however in a Spring application context we want it to be a managed bean.<\/p>\nFactoryBean<\/code><\/a> facility is clean and easy way to contain the details of making a singleton TwitterStream<\/code> instance.<\/strong><\/p>\npublic class TwitterStreamFactory extends AbstractFactoryBean<TwitterStream> {\n\n @Override\n public Class<?> getObjectType() {\n return TwitterStream.class;\n }\n\n @Override\n protected TwitterStream createInstance() {\n return new twitter4j.TwitterStreamFactory().getInstance();\n }\n\n @Override\n protected void destroyInstance(TwitterStream twitterStream) {\n twitterStream.shutdown();\n }\n\n}<\/pre>\nFactoryBean<\/code>, that wouldn\u2019t take care of properly shutting it down.<\/p>\n2.4.2. Attaching a listener and defining filtering<\/h2>\n
MessageProducer<\/code><\/a> implementation.<\/p>\n@Slf4j\npublic class TwitterMessageProducer extends MessageProducerSupport {\n\n private final TwitterStream twitterStream;\n\n private List<Long> follows;\n private List<String> terms;\n\n private StatusListener statusListener;\n private FilterQuery filterQuery;\n\n public TwitterMessageProducer(TwitterStream twitterStream, MessageChannel outputChannel) {\n this.twitterStream = twitterStream;\n setOutputChannel(outputChannel);\n }\n\n @Override\n protected void onInit() {\n super.onInit();\n\n statusListener = new StatusListener();\n\n long[] followsArray = null;\n\n if (!CollectionUtils.isEmpty(follows)) {\n followsArray = new long[follows.size()];\n for (int i = 0; i < follows.size(); i++) {\n followsArray[i] = follows.get(i);\n }\n }\n\n String[] termsArray = null;\n if (!CollectionUtils.isEmpty(terms)) {\n termsArray = terms.toArray(new String[0]);\n }\n\n filterQuery = new FilterQuery(0, followsArray, termsArray);\n }\n\n @Override\n public void doStart() {\n twitterStream.addListener(statusListener);\n twitterStream.filter(filterQuery);\n }\n\n @Override\n public void doStop() {\n twitterStream.cleanUp();\n twitterStream.clearListeners();\n }\n\n public void setFollows(List<Long> follows) {\n this.follows = follows;\n }\n\n public void setTerms(List<String> terms) {\n this.terms = terms;\n }\n\n StatusListener getStatusListener() {\n return statusListener;\n }\n\n FilterQuery getFilterQuery() {\n return filterQuery;\n }\n\n class StatusListener extends StatusAdapter {\n\n @Override\n public void onStatus(Status status) {\n sendMessage(MessageBuilder.withPayload(status).build());\n }\n\n @Override\n public void onException(Exception ex) {\n log.error(ex.getMessage(), ex);\n }\n\n @Override\n public void onStallWarning(StallWarning warning) {\n log.warn(warning.toString());\n }\n\n }\n}<\/pre>\nMessageProducerSupport<\/code> and TwitterStream<\/code>\u2018s management interface play nicely together. That\u2019s also going to enable us to stop and start the component at runtime when needed.<\/p>\n2.4.3. Java Configuration<\/h2>\n
@Slf4j\n@Configuration\npublic class TwitterConfig {\n\n @Bean\n TwitterStreamFactory twitterStreamFactory() {\n return new TwitterStreamFactory();\n }\n\n @Bean\n TwitterStream twitterStream(TwitterStreamFactory twitterStreamFactory) {\n return twitterStreamFactory.getInstance();\n }\n\n @Bean\n MessageChannel outputChannel() {\n return MessageChannels.direct().get();\n }\n\n @Bean\n TwitterMessageProducer twitterMessageProducer(\n TwitterStream twitterStream, MessageChannel outputChannel) {\n\n TwitterMessageProducer twitterMessageProducer =\n new TwitterMessageProducer(twitterStream, outputChannel);\n\n twitterMessageProducer.setTerms(Arrays.asList(\"java\", \"microservices\", \"spring\"));\n\n return twitterMessageProducer;\n }\n\n @Bean\n IntegrationFlow twitterFlow(MessageChannel outputChannel) {\n return IntegrationFlows.from(outputChannel)\n .transform(Status::getText)\n .handle(m -> log.info(m.getPayload().toString()))\n .get();\n }\n\n}<\/pre>\n2.5. Testing<\/h3>\n
@RunWith(SpringRunner.class)\n@ContextConfiguration(classes = TestConfig.class)\npublic class TwitterMessageProducerTest {\n\n @MockBean\n private TwitterStream twitterStream;\n\n @Autowired\n private PollableChannel outputChannel;\n\n @Autowired\n private TwitterMessageProducer twitterMessageProducer;\n\n @Test\n public void shouldBeInitialized() {\n StatusListener statusListener = twitterMessageProducer.getStatusListener();\n verify(twitterStream).addListener(statusListener);\n\n FilterQuery filterQuery = twitterMessageProducer.getFilterQuery();\n verify(twitterStream).filter(filterQuery);\n }\n\n @Test\n public void shouldReceiveStatus() {\n StatusListener statusListener = twitterMessageProducer.getStatusListener();\n\n Status status = mock(Status.class);\n statusListener.onStatus(status);\n\n Message<?> statusMessage = outputChannel.receive();\n assertSame(status, statusMessage.getPayload());\n }\n\n @Import(TwitterConfig.class)\n static class TestConfig {\n\n @Bean\n MessageChannel outputChannel() {\n return MessageChannels.queue(1).get();\n }\n\n }\n\n}<\/pre>\nTwitterStream<\/code> is no exception to that. That is, it can be mocked out easily in test cases.<\/strong><\/p>\n 6. Conclusion<\/h2>\n
\n