TwitterStreamSource implements Source { public void initialize(Properties properties) throws RequiredInputMissingException, ComponentInitializationFailedException { if(properties == null || properties.isEmpty()) throw new RequiredInputMissingException("Missing required configuration"); this.consumerKey = properties.getProperty(CFG_TWITTER_CONSUMER_KEY); this.consumerSecret = properties.getProperty(CFG_TWITTER_CONSUMER_SECRET); this.tokenKey = properties.getProperty(CFG_TWITTER_TOKEN_KEY); this.tokenSecret = properties.getProperty(CFG_TWITTER_TOKEN_SECRET); String inSearchTerms = properties.getProperty(CFG_TWITTER_TWEET_SEARCH_TERMS); String[] splittedSearchTerms = (inSearchTerms != null ? inSearchTerms.split(",") : null); if(splittedSearchTerms != null) { for(String sst : splittedSearchTerms) { this.searchTerms.add(StringUtils.trim(sst)); } } String inLanguages = properties.getProperty(CFG_TWITTER_TWEET_LANGUAGES); String[] splittedLanguages = (inLanguages != null ? inLanguages.split(",") : null); if(splittedLanguages != null) { for(String s : splittedLanguages) { this.languages.add(StringUtils.trim(s)); } } String inProfiles = properties.getProperty(CFG_TWITTER_PROFILES); String[] splittedProfiles = (inProfiles != null ? inProfiles.split(",") : null); if(splittedProfiles != null) { for(String sp : splittedProfiles) { if(StringUtils.isNotBlank(sp)) { try { this.profiles.add(Long.parseLong(sp.trim())); } catch(Exception e) { logger.error("Failed to parse profile identifier from input '"+sp+"'"); } } } } if(StringUtils.isBlank(id)) throw new RequiredInputMissingException("Missing required component identifier"); if(StringUtils.isBlank(this.consumerKey)) throw new RequiredInputMissingException("Missing required consumer key to establish connection with stream.twitter.com"); if(StringUtils.isBlank(this.consumerSecret)) throw new RequiredInputMissingException("Missing required consumer secrect to establish connection with stream.twitter.com"); if(StringUtils.isBlank(this.tokenKey)) throw new RequiredInputMissingException("Missing required token key to establish connection with stream.twitter.com"); if(StringUtils.isBlank(this.tokenSecret)) throw new RequiredInputMissingException("Missing required token secret to establish connection with stream.twitter.com"); boolean isFilterTermsEmpty = (this.searchTerms == null || this.searchTerms.isEmpty()); boolean isLanguagesEmpty = (this.languages == null || this.languages.isEmpty()); boolean isUserAccountEmpty = (this.profiles == null || this.profiles.isEmpty()); boolean isLocationsEmpty = (this.locations == null || this.locations.isEmpty()); if(isFilterTermsEmpty && isLanguagesEmpty && isUserAccountEmpty && isLocationsEmpty) throw new RequiredInputMissingException("Mishandle sing information what to filter twitter stream for: terms, languages, user accounts or locations"); Authentication auth = new OAuth1(this.consumerKey, this.consumerSecret, this.tokenKey, this.tokenSecret); StatusesFilterEndpoint filterEndpoint = new StatusesFilterEndpoint(); if(!isFilterTermsEmpty) filterEndpoint.trackTerms(searchTerms); if(!isLanguagesEmpty) filterEndpoint.languages(languages); if(!isUserAccountEmpty) filterEndpoint.followings(profiles); if(!isLocationsEmpty) filterEndpoint.locations(locations); if(this.twitterClient == null) { this.twitterClient = new ClientBuilder().name(id) .hosts(Constants.STREAM_HOST).endpoint(filterEndpoint).authentication(auth) .processor(new StringDelimitedProcessor(streamMessageQueue)).eventMessageQueue(eventMessageQueue).build(); this.twitterClient.connect(); } this.running = true; if(logger.isDebugEnabled()) logger.debug("twitter stream consumer initialized [id="+id+"]"); }  void initialize(Properties properties); boolean shutdown(); void run(); MicroPipelineComponentType getType(); void setIncomingMessageCallback(IncomingMessageCallback incomingMessageCallback); void setId(String id); String getId(); long getMessageCount(); static final String CFG_TWITTER_CONSUMER_KEY; static final String CFG_TWITTER_CONSUMER_SECRET; static final String CFG_TWITTER_TOKEN_KEY; static final String CFG_TWITTER_TOKEN_SECRET; static final String CFG_TWITTER_TWEET_SEARCH_TERMS; static final String CFG_TWITTER_TWEET_LANGUAGES; static final String CFG_TWITTER_PROFILES; }
@Test public void testInitialize_withNullInput() throws Exception { try { new TwitterStreamSource().initialize(null); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withEmptyProperties() throws Exception { try { new TwitterStreamSource().initialize(new Properties()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withEmptyComponentId() throws Exception { Properties props = new Properties(); props.put(TwitterStreamSource.CFG_TWITTER_CONSUMER_KEY, "ckey"); props.put(TwitterStreamSource.CFG_TWITTER_CONSUMER_SECRET, "csecret"); props.put(TwitterStreamSource.CFG_TWITTER_PROFILES, ""); props.put(TwitterStreamSource.CFG_TWITTER_TOKEN_KEY, "tkey"); props.put(TwitterStreamSource.CFG_TWITTER_TOKEN_SECRET, "tsecret"); props.put(TwitterStreamSource.CFG_TWITTER_TWEET_LANGUAGES, ""); props.put(TwitterStreamSource.CFG_TWITTER_TWEET_SEARCH_TERMS, ""); try { new TwitterStreamSource().initialize(props); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withEmptyConsumerKey() throws Exception { Properties props = new Properties(); props.put(TwitterStreamSource.CFG_TWITTER_CONSUMER_KEY, ""); props.put(TwitterStreamSource.CFG_TWITTER_CONSUMER_SECRET, "csecret"); props.put(TwitterStreamSource.CFG_TWITTER_PROFILES, ""); props.put(TwitterStreamSource.CFG_TWITTER_TOKEN_KEY, "tkey"); props.put(TwitterStreamSource.CFG_TWITTER_TOKEN_SECRET, "tsecret"); props.put(TwitterStreamSource.CFG_TWITTER_TWEET_LANGUAGES, ""); props.put(TwitterStreamSource.CFG_TWITTER_TWEET_SEARCH_TERMS, ""); try { new TwitterStreamSource().initialize(props); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withEmptyConsumerSecret() throws Exception { Properties props = new Properties(); props.put(TwitterStreamSource.CFG_TWITTER_CONSUMER_KEY, "ckey"); props.put(TwitterStreamSource.CFG_TWITTER_CONSUMER_SECRET, ""); props.put(TwitterStreamSource.CFG_TWITTER_PROFILES, ""); props.put(TwitterStreamSource.CFG_TWITTER_TOKEN_KEY, "tkey"); props.put(TwitterStreamSource.CFG_TWITTER_TOKEN_SECRET, "tsecret"); props.put(TwitterStreamSource.CFG_TWITTER_TWEET_LANGUAGES, ""); props.put(TwitterStreamSource.CFG_TWITTER_TWEET_SEARCH_TERMS, ""); try { new TwitterStreamSource().initialize(props); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withEmptyTokenKey() throws Exception { Properties props = new Properties(); props.put(TwitterStreamSource.CFG_TWITTER_CONSUMER_KEY, "ckey"); props.put(TwitterStreamSource.CFG_TWITTER_CONSUMER_SECRET, "csecret"); props.put(TwitterStreamSource.CFG_TWITTER_PROFILES, ""); props.put(TwitterStreamSource.CFG_TWITTER_TOKEN_KEY, ""); props.put(TwitterStreamSource.CFG_TWITTER_TOKEN_SECRET, "tsecret"); props.put(TwitterStreamSource.CFG_TWITTER_TWEET_LANGUAGES, ""); props.put(TwitterStreamSource.CFG_TWITTER_TWEET_SEARCH_TERMS, ""); try { new TwitterStreamSource().initialize(props); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitialize_withEmptyTokenSecret() throws Exception { Properties props = new Properties(); props.put(TwitterStreamSource.CFG_TWITTER_CONSUMER_KEY, "ckey"); props.put(TwitterStreamSource.CFG_TWITTER_CONSUMER_SECRET, "csecret"); props.put(TwitterStreamSource.CFG_TWITTER_PROFILES, ""); props.put(TwitterStreamSource.CFG_TWITTER_TOKEN_KEY, "tkey"); props.put(TwitterStreamSource.CFG_TWITTER_TOKEN_SECRET, ""); props.put(TwitterStreamSource.CFG_TWITTER_TWEET_LANGUAGES, ""); props.put(TwitterStreamSource.CFG_TWITTER_TWEET_SEARCH_TERMS, ""); try { new TwitterStreamSource().initialize(props); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }