@Test @SuppressWarnings("unchecked") public void getHosebirdCient_should_create_client_correctly() { Producer<String, String> producer = mock(Producer.class); GnipReader tr = mock(GnipReader.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("Gnip Reader"); verify(cb).hosts(Constants.ENTERPRISE_STREAM_HOST_v2); ArgumentCaptor<StreamingEndpoint> se = ArgumentCaptor.forClass(StreamingEndpoint.class); verify(cb).endpoint(se.capture()); assertEquals("/stream/PRODUCT/accounts/ACCOUNT/publishers/twitter/LABEL.json", se.getValue().getURI()); ArgumentCaptor<Authentication> a = ArgumentCaptor.forClass(Authentication.class); verify(cb).authentication(a.capture()); BasicAuth expectedAuth = new BasicAuth("USERNAME", "PASSWORD"); ArgumentCaptor<HosebirdMessageProcessor> hbmp = ArgumentCaptor.forClass(HosebirdMessageProcessor.class); verify(cb).processor(hbmp.capture()); LineStringProcessor expectedProcessor = new LineStringProcessor(lbq); verify(cb).retries(10); }
@SuppressWarnings("checkstyle:designforextension") @Override protected Client getHosebirdClient( final LinkedBlockingQueue<String> buffer, final Config readerConfig) { getLogger().info("Building Gnip client"); GnipConfig config = ((GnipReaderConfig) readerConfig).gnip; RealTimeEnterpriseStreamingEndpoint_v2 endpoint = new RealTimeEnterpriseStreamingEndpoint_v2( config.account, config.product, config.label); BasicAuth auth = new BasicAuth(config.username, config.password); LineStringProcessor processor = new LineStringProcessor(buffer, readerConfig.hosebird.bufferTimeout); return this.getClientBuilder() .name("Gnip Reader") .hosts(config.host) .endpoint(endpoint) .authentication(auth) .processor(processor) .retries(readerConfig.hosebird.retries) .build(); }
GnipReader extends HosebirdReader { @SuppressWarnings("checkstyle:designforextension") @Override protected Client getHosebirdClient( final LinkedBlockingQueue<String> buffer, final Config readerConfig) { getLogger().info("Building Gnip client"); GnipConfig config = ((GnipReaderConfig) readerConfig).gnip; RealTimeEnterpriseStreamingEndpoint_v2 endpoint = new RealTimeEnterpriseStreamingEndpoint_v2( config.account, config.product, config.label); BasicAuth auth = new BasicAuth(config.username, config.password); LineStringProcessor processor = new LineStringProcessor(buffer, readerConfig.hosebird.bufferTimeout); return this.getClientBuilder() .name("Gnip Reader") .hosts(config.host) .endpoint(endpoint) .authentication(auth) .processor(processor) .retries(readerConfig.hosebird.retries) .build(); } }
GnipReader extends HosebirdReader { @SuppressWarnings("checkstyle:designforextension") @Override protected Client getHosebirdClient( final LinkedBlockingQueue<String> buffer, final Config readerConfig) { getLogger().info("Building Gnip client"); GnipConfig config = ((GnipReaderConfig) readerConfig).gnip; RealTimeEnterpriseStreamingEndpoint_v2 endpoint = new RealTimeEnterpriseStreamingEndpoint_v2( config.account, config.product, config.label); BasicAuth auth = new BasicAuth(config.username, config.password); LineStringProcessor processor = new LineStringProcessor(buffer, readerConfig.hosebird.bufferTimeout); return this.getClientBuilder() .name("Gnip Reader") .hosts(config.host) .endpoint(endpoint) .authentication(auth) .processor(processor) .retries(readerConfig.hosebird.retries) .build(); } GnipReader(); }
GnipReader extends HosebirdReader { @SuppressWarnings("checkstyle:designforextension") @Override protected Client getHosebirdClient( final LinkedBlockingQueue<String> buffer, final Config readerConfig) { getLogger().info("Building Gnip client"); GnipConfig config = ((GnipReaderConfig) readerConfig).gnip; RealTimeEnterpriseStreamingEndpoint_v2 endpoint = new RealTimeEnterpriseStreamingEndpoint_v2( config.account, config.product, config.label); BasicAuth auth = new BasicAuth(config.username, config.password); LineStringProcessor processor = new LineStringProcessor(buffer, readerConfig.hosebird.bufferTimeout); return this.getClientBuilder() .name("Gnip Reader") .hosts(config.host) .endpoint(endpoint) .authentication(auth) .processor(processor) .retries(readerConfig.hosebird.retries) .build(); } GnipReader(); static void main(final String[] args); }
GnipReader extends HosebirdReader { @SuppressWarnings("checkstyle:designforextension") @Override protected Client getHosebirdClient( final LinkedBlockingQueue<String> buffer, final Config readerConfig) { getLogger().info("Building Gnip client"); GnipConfig config = ((GnipReaderConfig) readerConfig).gnip; RealTimeEnterpriseStreamingEndpoint_v2 endpoint = new RealTimeEnterpriseStreamingEndpoint_v2( config.account, config.product, config.label); BasicAuth auth = new BasicAuth(config.username, config.password); LineStringProcessor processor = new LineStringProcessor(buffer, readerConfig.hosebird.bufferTimeout); return this.getClientBuilder() .name("Gnip Reader") .hosts(config.host) .endpoint(endpoint) .authentication(auth) .processor(processor) .retries(readerConfig.hosebird.retries) .build(); } GnipReader(); static void main(final String[] args);  }