import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
private static BasicClient configureStreamClient(BlockingQueue<String> msgQueue, String twitterKeys, List<Long> userIds, List<String> terms) {
    Hosts hosts = new HttpHosts(Constants.STREAM_HOST);
    StatusesFilterEndpoint endpoint = new StatusesFilterEndpoint()
            .followings(userIds)
            .trackTerms(terms);
    endpoint.stallWarnings(false);

    String[] keys = twitterKeys.split(":");
    Authentication auth = new OAuth1(keys[0], keys[1], keys[2], keys[3]);

    ClientBuilder builder = new ClientBuilder()
            .name("Neo4j-Twitter-Stream")
            .hosts(hosts)
            .authentication(auth)
            .endpoint(endpoint)
            .processor(new StringDelimitedProcessor(msgQueue));

    return builder.build();
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
private  Client getTwitterClient(Properties props, BlockingQueue<String> messageQueue) {

        String clientName = props.getProperty("clientName");
        String consumerKey = props.getProperty("consumerKey");
        String consumerSecret = props.getProperty("consumerSecret");
        String token = props.getProperty("token");
        String tokenSecret = props.getProperty("tokenSecret");
        List<String> searchTerms = Arrays.asList(props.getProperty("searchTerms").split(","));

        Authentication authentication = new OAuth1(consumerKey,consumerSecret,token,tokenSecret);
        Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
        StatusesFilterEndpoint hosebirdEndpoint = new StatusesFilterEndpoint();

        hosebirdEndpoint.trackTerms(searchTerms);

        ClientBuilder clientBuilder = new ClientBuilder();
        clientBuilder.name(clientName)
                .hosts(hosebirdHosts)
                .authentication(authentication)
                .endpoint(hosebirdEndpoint)
                .processor(new StringDelimitedProcessor(messageQueue));

          return clientBuilder.build();

    }
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
/**
 * @throws Exception Test exception.
 */
public void testStatusesFilterEndpointOAuth1() throws Exception {
    try (IgniteDataStreamer<Long, String> dataStreamer = grid().dataStreamer(DEFAULT_CACHE_NAME)) {
        TwitterStreamerImpl streamer = newStreamerInstance(dataStreamer);

        Map<String, String> params = new HashMap<>();

        params.put("track", "apache, twitter");
        params.put("follow", "3004445758");// @ApacheIgnite id.

        streamer.setApiParams(params);
        streamer.setEndpointUrl(MOCK_TWEET_PATH);
        streamer.setHosts(new HttpHosts("http://localhost:8080"));
        streamer.setThreadsCount(8);

        executeStreamer(streamer);
    }
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
public static void setupHosebirdClient() {
       /** Declare the host you want to connect to, the endpoint, and authentication (basic auth or oauth) */
	Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
	StatusesFilterEndpoint endpoint = new StatusesFilterEndpoint();

       // Optional: set up some followings and track terms
	List<Long> followings = Lists.newArrayList(1234L, 566788L);
	List<String> terms = Lists.newArrayList("twitter", "api");
	endpoint.followings(followings);
	endpoint.trackTerms(terms);

	Authentication hosebirdAuth = new OAuth1(
       		Helper.properties().getProperty("consumerKey"),
       		Helper.properties().getProperty("consumerSecret"),
       		Helper.properties().getProperty("token"),
       		Helper.properties().getProperty("secret"));

       ClientBuilder builder = new ClientBuilder()
        .name("Hosebird-Client-01")		// optional: mainly for the logs
        .hosts(hosebirdHosts)
        .authentication(hosebirdAuth)
        .endpoint(endpoint)
        .processor(new StringDelimitedProcessor(msgQueue));

	hosebirdClient = builder.build();
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testProperlyHandleSuccessfulConnection() {
  ClientBase clientBase = new ClientBase("name",
          mock, new HttpHosts("http://hi"), new RawEndpoint("/endpoint", HttpConstants.HTTP_GET), mockAuth,
          mockProcessor, mockReconnectionManager, mockRateTracker
  );

  mockStatusLine = mock(StatusLine.class);
  when(mockStatusLine.getStatusCode())
          .thenReturn(HttpConstants.Codes.SUCCESS);

  assertTrue(clientBase.handleConnectionResult(mockStatusLine));

  InOrder inOrder = inOrder(mockStatusLine, mockReconnectionManager);
  inOrder.verify(mockStatusLine).getStatusCode();
  inOrder.verify(mockReconnectionManager).resetCounts();

  assertFalse(clientBase.isDone());
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testHandleIOExceptionOnConnection() throws IOException {
  ClientBase clientBase = new ClientBase("name",
          mock, new HttpHosts("http://hi"), new RawEndpoint("/endpoint", HttpConstants.HTTP_GET), mockAuth,
          mockProcessor, mockReconnectionManager, mockRateTracker
  );
  when(mockConnection.connect(any(HttpUriRequest.class)))
          .thenThrow(new IOException());

  HttpUriRequest mockRequest = mock(HttpUriRequest.class);
  assertNull(clientBase.establishConnection(mockConnection, mockRequest));

  InOrder inOrder = inOrder(mockConnection, mockReconnectionManager);

  inOrder.verify(mockConnection).connect(any(HttpUriRequest.class));
  inOrder.verify(mockReconnectionManager).handleLinearBackoff();

  assertFalse(clientBase.isDone());
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testRetryTransientAuthFailures() {
  ClientBase clientBase = new ClientBase("name",
          mock, new HttpHosts("http://hi"), new RawEndpoint("/endpoint", HttpConstants.HTTP_GET), mockAuth,
          mockProcessor, mockReconnectionManager, mockRateTracker
  );

  when(mockStatusLine.getStatusCode())
          .thenReturn(401);
  when(mockStatusLine.getReasonPhrase())
          .thenReturn("reason");
  when(mockReconnectionManager.shouldReconnectOn400s())
          .thenReturn(true, true, false);

  // auth failure 3 times. We'll retry the first two times, but give up on the 3rd
  clientBase.handleConnectionResult(mockStatusLine);
  clientBase.handleConnectionResult(mockStatusLine);
  verify(mockReconnectionManager, times(2)).handleExponentialBackoff();
  assertFalse(clientBase.isDone());
  clientBase.handleConnectionResult(mockStatusLine);
  verify(mockReconnectionManager, times(2)).handleExponentialBackoff();
  assertTrue(clientBase.isDone());
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testServiceUnavailable() {
  ClientBase clientBase = new ClientBase("name",
          mock, new HttpHosts("http://hi"), new RawEndpoint("/endpoint", HttpConstants.HTTP_GET), mockAuth,
          mockProcessor, mockReconnectionManager, mockRateTracker
  );
  when(mockStatusLine.getStatusCode())
          .thenReturn(503);

  clientBase.handleConnectionResult(mockStatusLine);
  clientBase.handleConnectionResult(mockStatusLine);
  clientBase.handleConnectionResult(mockStatusLine);
  clientBase.handleConnectionResult(mockStatusLine);

  verify(mockReconnectionManager, times(4)).handleExponentialBackoff();
  assertFalse(clientBase.isDone());
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testInterruptedExceptionDuringProcessing() throws Exception {
  ClientBase clientBase = new ClientBase("name",
          mockClient, new HttpHosts("http://hi"), new RawEndpoint("/endpoint", HttpConstants.HTTP_GET), mockAuth,
          mockProcessor, mockReconnectionManager, mockRateTracker
  );

  when(mockStatusLine.getStatusCode())
          .thenReturn(200);

  doThrow(new InterruptedException()).when(mockProcessor).process();

  when(mockClient.getConnectionManager())
          .thenReturn(mockConnectionManager);

  BasicClient client = new BasicClient(clientBase, executorService);

  assertFalse(clientBase.isDone());
  client.connect();
  assertTrue(client.waitForFinish(100));
  assertTrue(client.isDone());
  verify(mockProcessor).setup(any(InputStream.class));
  verify(mockConnectionManager, atLeastOnce()).shutdown();
  assertEquals(EventType.STOPPED_BY_ERROR, client.getExitEvent().getEventType());
  assertTrue(client.getExitEvent().getUnderlyingException() instanceof InterruptedException);
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testExceptionIfBadScheme() {
  int size = 50;
  List<String> hosts = new ArrayList<String>(size);
  for (int i = 0; i < size; i++) {
    if (i == size / 2) {
      hosts.add("thisfails.com");
    } else {
      hosts.add("https://thisismyawesomehost " + i + ".com");
    }
  }
  try {
    new HttpHosts(hosts);
    fail();
  } catch (RuntimeException e) {
    // is expected
  }
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testIsScrambled() {
  int size = 50;
  List<String> hosts = new ArrayList<String>(size);
  for (int i = 0; i < size; i++) {
    hosts.add("http://thisismyawesomehost " + i + ".com");
  }
  HttpHosts httpHosts = new HttpHosts(hosts);

  boolean allSame = true;
  for (String string : hosts) {
    if (!string.equals(httpHosts.nextHost())) {
      allSame = false;
      break;
    }
  }
  assertFalse("This test failed unless you got EXTREMELY unlucky.", allSame);
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
/**
 * @param tweetQueue tweet Queue.
 * @param hosts Hostes.
 * @param endpoint Endpoint.
 * @return Client.
 */
protected Client buildClient(BlockingQueue<String> tweetQueue, HttpHosts hosts, StreamingEndpoint endpoint) {
    Authentication authentication = new OAuth1(oAuthSettings.getConsumerKey(), oAuthSettings.getConsumerSecret(),
        oAuthSettings.getAccessToken(), oAuthSettings.getAccessTokenSecret());

    ClientBuilder builder = new ClientBuilder()
        .name("Ignite-Twitter-Client")
        .hosts(hosts)
        .authentication(authentication)
        .endpoint(endpoint)
        .processor(new StringDelimitedProcessor(tweetQueue));

    return builder.build();
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
public void subscribe(final StatusStreamHandler listener, String... terms) {
	/**
	 * Set up your blocking queues: Be sure to size these properly based on
	 * expected TPS of your stream
	 */
	BlockingQueue<String> msgQueue = new LinkedBlockingQueue<String>(100000);
	BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<Event>(1000);

	/**
	 * Declare the host you want to connect to, the endpoint, and
	 * authentication (basic auth or oauth)
	 */
	Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
	StatusesFilterEndpoint hosebirdEndpoint = new StatusesFilterEndpoint();
	hosebirdEndpoint.trackTerms(Lists.newArrayList(terms));

	Authentication hosebirdAuth = oAuth();

	ClientBuilder builder = new ClientBuilder().name("Hosebird-Client-01")
			// optional: mainly for the logs
			.hosts(hosebirdHosts).authentication(hosebirdAuth).endpoint(hosebirdEndpoint)
			.processor(new StringDelimitedProcessor(msgQueue)).eventMessageQueue(eventQueue);
	Client client = builder.build();

	final ExecutorService executorService = Executors.newFixedThreadPool(1);
	final Twitter4jStatusClient t4jClient = new Twitter4jStatusClient(client, msgQueue,
			Lists.newArrayList(listener), executorService);
	t4jClient.connect();

	// Call this once for every thread you want to spin off for processing
	// the raw messages.
	// This should be called at least once.
	t4jClient.process(); // required to start processing the messages
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testUnknownEndpointFails() {
  ClientBase clientBase = new ClientBase("name",
          mock, new HttpHosts("http://hi"), new RawEndpoint("/endpoint", HttpConstants.HTTP_GET), mockAuth,
          mockProcessor, mockReconnectionManager, mockRateTracker
  );
  when(mockStatusLine.getStatusCode())
          .thenReturn(404);
  when(mockStatusLine.getReasonPhrase())
          .thenReturn("reason");
  clientBase.handleConnectionResult(mockStatusLine);
  assertTrue(clientBase.isDone());
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testIOExceptionDuringProcessing() throws Exception {
  ClientBase clientBase = new ClientBase("name",
          mockClient, new HttpHosts("http://hi"), new RawEndpoint("/endpoint", HttpConstants.HTTP_GET), mockAuth,
          mockProcessor, mockReconnectionManager, mockRateTracker
  );
  BasicClient client = new BasicClient(clientBase, executorService);
  final CountDownLatch latch = new CountDownLatch(1);
  when(mockStatusLine.getStatusCode())
          .thenReturn(200);

  doNothing().when(mockProcessor).setup(any(InputStream.class));
  doThrow(new IOException()).
          doThrow(new IOException()).
          doThrow(new IOException()).
          doAnswer(new Answer() {
            @Override
            public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
              latch.countDown();
              return null;
            }
          }).when(mockProcessor).process();

  client.connect();
  latch.await();
  assertFalse(clientBase.isDone());
  verify(mockProcessor, times(4)).setup(any(InputStream.class));
  // throw 3 exceptions, 4th one keeps going
  verify(mockProcessor, atLeast(4)).process();

  client.stop();
  verify(mockConnectionManager, atLeastOnce()).shutdown();
  assertTrue(client.isDone());
  assertEquals(EventType.STOPPED_BY_USER, clientBase.getExitEvent().getEventType());
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testBuilderSuccess() {
  new ClientBuilder()
          .hosts(new HttpHosts(Constants.STREAM_HOST))
          .endpoint(new StatusesSampleEndpoint())
          .processor(new NullProcessor())
          .authentication(new BasicAuth("username", "password"))
          .build();

}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testInvalidHttpMethod() {
  try {
    new ClientBuilder()
            .hosts(new HttpHosts(Constants.STREAM_HOST))
            .endpoint(StatusesSampleEndpoint.PATH, "FAIL!")
            .processor(new NullProcessor())
            .authentication(new BasicAuth("username", "password"))
            .build();
    fail();
  } catch (Exception e) {
    // expected
  }
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testValidHttpMethod() {
  new ClientBuilder()
          .hosts(new HttpHosts(Constants.STREAM_HOST))
          .endpoint(StatusesSampleEndpoint.PATH, "gEt")
          .processor(new NullProcessor())
          .authentication(new BasicAuth("username", "password"))
          .build();

}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testContainsAll() {
  int size = 30;
  List<String> hosts = new ArrayList<String>(size);
  for (int i = 0; i < size; i++) {
    hosts.add("http://thisismyawesomehost " + i + ".com");
  }
  HttpHosts httpHosts = new HttpHosts(hosts);
  Set<String> hostsSet = new HashSet<String>(hosts);
  for (int i = 0; i < size; i++) {
    assertTrue(hostsSet.remove(httpHosts.nextHost()));
  }
  assertTrue(hostsSet.isEmpty());
}
 

import com.twitter.hbc.core.HttpHosts; //导入依赖的package包/类
@Test
public void testInfiniteIteration() {
  int size = 10;
  List<String> hosts = new ArrayList<String>(size);
  for (int i = 0; i < size; i++) {
    hosts.add("http://thisismyawesomehost " + i + ".com");
  }
  HttpHosts httpHosts = new HttpHosts(hosts);
  Set<String> hostsSet = new HashSet<String>(hosts);
  for (int i = 0; i < size * 10; i++) {
    assertTrue(hostsSet.contains(httpHosts.nextHost()));
  }
}