TwitterApiReader extends HosebirdReader { @SuppressWarnings("checkstyle:designforextension") @Override protected Client getHosebirdClient( final LinkedBlockingQueue<String> buffer, final Config readerConfig) { getLogger().info("Building Twitter client"); StatusesFilterEndpoint endpoint = new StatusesFilterEndpoint(); TwitterApiConfig config = ((TwitterApiReaderConfig) readerConfig).twitterapi; if (config.keywords != null) { endpoint.trackTerms(config.keywords); } if (config.userIds != null) { endpoint.followings(config.userIds); } Authentication auth = new OAuth1( config.consumerKey, config.consumerSecret, config.accessToken, config.accessSecret ); LineStringProcessor processor = new LineStringProcessor(buffer, readerConfig.hosebird.bufferTimeout); return this.getClientBuilder() .name("Twitter Api Reader") .hosts(Constants.STREAM_HOST) .endpoint(endpoint) .authentication(auth) .retries(readerConfig.hosebird.retries) .processor(processor) .build(); } TwitterApiReader(); static void main(final String[] args);  }
@Test @SuppressWarnings("unchecked") public void getHosebirdCient_should_create_client_correctly() { Producer<String, String> producer = mock(Producer.class); TwitterApiReader tr = mock(TwitterApiReader.class); reset(this.logger); tr.setLogger(this.logger); when(tr.getMetrics(any(StatsReporter.StatsTracker.class))).thenCallRealMethod(); when(tr.getHosebirdClient(any(LinkedBlockingQueue.class), any(Config.class))).thenCallRealMethod(); when(tr.getKafkaProducer(any(KafkaConfig.class))).thenReturn(producer); when(tr.getClientBuilder()).thenReturn(this.cb); when(tr.parseConfigFile(anyString())).thenReturn(config); when(tr.getRetry()).thenReturn(true).thenReturn(false); when(client.isDone()).thenReturn(true); LinkedBlockingQueue<String> lbq = new LinkedBlockingQueue<>(10); tr.getHosebirdClient(lbq, config); verify(cb).name("Twitter Api Reader"); verify(cb).hosts(Constants.STREAM_HOST); ArgumentCaptor<StreamingEndpoint> se = ArgumentCaptor.forClass(StreamingEndpoint.class); verify(cb).endpoint(se.capture()); assertEquals("/1.1/statuses/filter.json?delimited=length&stall_warnings=true", se.getValue().getURI()); verify(cb).retries(10); }