TwitterConnector { public TwitterConnector(List<String> keywordList, List<Location> locationList, List<String> languageList, String customerKey, String customerSecret, String token, String tokenSecret) throws TexeraException { if ((keywordList == null || keywordList.isEmpty()) && (locationList == null || locationList.isEmpty()) && (languageList == null || languageList.isEmpty())) { throw new DataflowException("no filter is provided"); } messageQueue = new LinkedBlockingQueue<String>(queueSize); StatusesFilterEndpoint endpoint = new StatusesFilterEndpoint(); if (!(keywordList == null || keywordList.isEmpty())) { endpoint.trackTerms(keywordList); } if (!(locationList == null || locationList.isEmpty())) { endpoint.locations(locationList); } if (!(languageList == null || languageList.isEmpty())) { endpoint.languages(languageList); } if ((customerKey != null && !customerKey.isEmpty()) && (customerSecret != null && !customerSecret.isEmpty()) && (token != null && !token.isEmpty()) && (tokenSecret != null && !tokenSecret.isEmpty())) { auth = new OAuth1(customerKey, customerSecret, token, tokenSecret); } else { auth = new OAuth1(TwitterUtils.twitterConstant.consumerKey, TwitterUtils.twitterConstant.consumerSecret, TwitterUtils.twitterConstant.token, TwitterUtils.twitterConstant.tokenSecret); } twitterClient = new ClientBuilder() .hosts(Constants.STREAM_HOST) .endpoint(endpoint) .authentication(auth) .processor(new StringDelimitedProcessor(messageQueue)) .build(); } TwitterConnector(List<String> keywordList, List<Location> locationList, List<String> languageList, String customerKey, String customerSecret, String token, String tokenSecret); BasicClient getClient(); void setMessageQueue(BlockingQueue<String> queue); BlockingQueue<String> getMsgQueue();  }
@Test public void testTwitterConnector() throws Exception { int tweetCount = 0; int tweetNumberToAccess = 2; while (tweetCount < tweetNumberToAccess) { String message = twitterConnector.getMsgQueue().take(); assertTrue(!TwitterUtils.getUserScreenName(new ObjectMapper().readValue(message, JsonNode.class)).isEmpty()); tweetCount++; } }