import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
@Test
public void should_log_reason_that_client_stopped() {
    Event event = new Event(EventType.STOPPED_BY_ERROR, new Exception("EXCEPTION"));
    when(this.client.getExitEvent()).thenReturn(event);

    HosebirdReader tr = new ConcreteHosebirdReader();
    reset(this.logger);
    tr.setLogger(this.logger);

    tr.logClientExitReason(this.client);

    verify(this.logger).error(
            "Hosebird client stopped: {} {}",
            new Object[]{
                    "STOPPED_BY_ERROR",
                    "EXCEPTION"
            });
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
ClientBase(String name, HttpClient client, Hosts hosts, StreamingEndpoint endpoint, Authentication auth,
           HosebirdMessageProcessor processor, ReconnectionManager manager, RateTracker rateTracker,
           @Nullable BlockingQueue<Event> eventsQueue) {
    this.client = Preconditions.checkNotNull(client);
    this.name = Preconditions.checkNotNull(name);

    this.endpoint = Preconditions.checkNotNull(endpoint);
    this.hosts = Preconditions.checkNotNull(hosts);
    this.auth = Preconditions.checkNotNull(auth);

    this.processor = Preconditions.checkNotNull(processor);
    this.reconnectionManager = Preconditions.checkNotNull(manager);
    this.rateTracker = Preconditions.checkNotNull(rateTracker);

    this.eventsQueue = eventsQueue;

    this.exitEvent = new AtomicReference<Event>();

    this.isRunning = new CountDownLatch(1);
    this.statsReporter = new StatsReporter();

    this.connectionEstablished = new AtomicBoolean(false);
    this.reconnect = new AtomicBoolean(false);
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
public BasicClient(String name, Hosts hosts, StreamingEndpoint endpoint, Authentication auth, boolean enableGZip, HosebirdMessageProcessor processor,
                   ReconnectionManager reconnectionManager, RateTracker rateTracker, ExecutorService executorService,
                   @Nullable BlockingQueue<Event> eventsQueue, HttpParams params, SchemeRegistry schemeRegistry) {
  Preconditions.checkNotNull(auth);
  HttpClient client;
  if (enableGZip) {
    client = new RestartableHttpClient(auth, enableGZip, params, schemeRegistry);
  } else {
    DefaultHttpClient defaultClient = new DefaultHttpClient(new PoolingClientConnectionManager(schemeRegistry), params);

    /** Set auth **/
    auth.setupConnection(defaultClient);
    client = defaultClient;
  }

  this.canRun = new AtomicBoolean(true);
  this.executorService = executorService;
  this.clientBase = new ClientBase(name, client, hosts, endpoint, auth, processor, reconnectionManager, rateTracker, eventsQueue);
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
ClientBase(String name, HttpClient client, Hosts hosts, StreamingEndpoint endpoint, Authentication auth,
           HosebirdMessageProcessor processor, ReconnectionManager manager, RateTracker rateTracker,
           @Nullable BlockingQueue<Event> eventsQueue) {
  this.client = Preconditions.checkNotNull(client);
  this.name = Preconditions.checkNotNull(name);

  this.endpoint = Preconditions.checkNotNull(endpoint);
  this.hosts = Preconditions.checkNotNull(hosts);
  this.auth = Preconditions.checkNotNull(auth);

  this.processor = Preconditions.checkNotNull(processor);
  this.reconnectionManager = Preconditions.checkNotNull(manager);
  this.rateTracker = Preconditions.checkNotNull(rateTracker);

  this.eventsQueue = eventsQueue;

  this.exitEvent = new AtomicReference<Event>();

  this.isRunning = new CountDownLatch(1);
  this.statsReporter = new StatsReporter();

  this.connectionEstablished = new AtomicBoolean(false);
  this.reconnect = new AtomicBoolean(false);
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
/**
 * Logs the reason that the Hosebird client said it was done
 * by examining the exit event.
 * @param client the Hosebird client
 */
@VisibleForTesting
@SuppressWarnings("checkstyle:designforextension")
protected void logClientExitReason(final Client client) {
    BasicClient bc = (BasicClient) client;
    Event e = bc.getExitEvent();

    log.error(
            "Hosebird client stopped: {} {}",
            new Object[]{
            e.getEventType().name(),
            e.getMessage()});
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
private void addEvent(Event event) {
    if (eventsQueue != null) {
        if (!eventsQueue.offer(event)) {
            statsReporter.incrNumClientEventsDropped();
        }
    }
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
/**
 * Stops the current connection. No reconnecting will occur. Kills thread + cleanup.
 * Waits for the loop to end
 **/
public void stop(int waitMillis) throws InterruptedException {
    try {
        if (!isDone()) {
            setExitStatus(new Event(EventType.STOPPED_BY_USER, String.format("Stopped by user: waiting for %d ms", waitMillis)));
        }
        if (!waitForFinish(waitMillis)) {
            logger.warn("{} Client thread failed to finish in {} millis", name, waitMillis);
        }
    } finally {
        rateTracker.shutdown();
    }
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的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.event.Event; //导入依赖的package包/类
private void addEvent(Event event) {
  if (eventsQueue != null) {
    if (!eventsQueue.offer(event)) {
      statsReporter.incrNumClientEventsDropped();
    }
  }
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
/**
 * Stops the current connection. No reconnecting will occur. Kills thread + cleanup.
 * Waits for the loop to end
 **/
public void stop(int waitMillis) throws InterruptedException {
  try {
    if (!isDone()) {
      setExitStatus(new Event(EventType.STOPPED_BY_USER, String.format("Stopped by user: waiting for %d ms", waitMillis)));
    }
    if (!waitForFinish(waitMillis)) {
      logger.warn("{} Client thread failed to finish in {} millis", name, waitMillis);
    }
  } finally {
    rateTracker.shutdown();
  }
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
public TwitterClient() {
	/** 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();
	// Optional: set up some followings and track terms
	List<Long> followings = Lists.newArrayList(1234L, 566788L);
	List<String> terms = Lists.newArrayList("twitter", "api");
	hosebirdEndpoint.followings(followings);
	hosebirdEndpoint.trackTerms(terms);

	// These secrets should be read from a config file
	Authentication hosebirdAuth = new OAuth1("consumerKey", "consumerSecret", "token", "secret");

	ClientBuilder builder = new ClientBuilder()
	  .name("Hosebird-Client-01")                              // optional: mainly for the logs
	  .hosts(hosebirdHosts)
	  .authentication(hosebirdAuth)
	  .endpoint(new StatusesSampleEndpoint())
	  .processor(new StringDelimitedProcessor(msgQueue))
	  .eventMessageQueue(eventQueue);                          // optional: use this if you want to process client events

	Client hosebirdClient = builder.build();
	// Attempts to establish a connection.
	hosebirdClient.connect();
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
@Override
public void onTrigger(final ProcessContext context,
		final ProcessSession session) throws ProcessException {
	final Event event = eventQueue.poll();
	if (event != null) {
		switch (event.getEventType()) {
		case STOPPED_BY_ERROR:
			getLogger()
					.error("Received error {}: {} due to {}. Will not attempt to reconnect",
							new Object[] { event.getEventType(),
									event.getMessage(),
									event.getUnderlyingException() });
			break;
		case CONNECTION_ERROR:
		case HTTP_ERROR:
			getLogger()
					.error("Received error {}: {}. Will attempt to reconnect",
							new Object[] { event.getEventType(),
									event.getMessage() });
			client.reconnect();
			break;
		default:
			break;
		}
	}

	final String tweet = messageQueue.poll();
	if (tweet == null) {
		context.yield();
		return;
	}

	FlowFile flowFile = session.create();
	flowFile = session.write(flowFile, new OutputStreamCallback() {
		@Override
		public void process(final OutputStream out) throws IOException {
			out.write(tweet.getBytes(StandardCharsets.UTF_8));
		}
	});

	final Map<String, String> attributes = new HashMap<>();
	attributes.put(CoreAttributes.MIME_TYPE.key(), "application/json");
	attributes.put(CoreAttributes.FILENAME.key(),
			flowFile.getAttribute(CoreAttributes.FILENAME.key()) + ".json");
	flowFile = session.putAllAttributes(flowFile, attributes);

	session.transfer(flowFile, REL_SUCCESS);
	session.getProvenanceReporter().receive(
			flowFile,
			Constants.STREAM_HOST
					+ client.getEndpoint().getURI().toString());
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
@Override
public void run() {
    // establish the initial connection
    //   if connection fails due to auth or some other 400, stop immediately
    //   if connection fails due to a 500, back off and retry
    //   if no response or other code, stop immediately
    // begin reading from the stream
    // while the stop signal hasn't been sent, and no IOException from processor, keep processing
    // if  IOException, time to restart the connection:
    //   handle http connection cleanup
    //   do some backoff, set backfill
    // if stop signal set, time to kill/clean the connection, and end this thread.
    try {
        if (client instanceof RestartableHttpClient) {
            ((RestartableHttpClient) client).setup();
        }
        rateTracker.start();
        while (!isDone()) {
            String host = hosts.nextHost();
            if (host == null) {
                setExitStatus(new Event(EventType.STOPPED_BY_ERROR, "No hosts available"));
                break;
            }

            double rate = rateTracker.getCurrentRateSeconds();
            if (!Double.isNaN(rate)) {
                endpoint.setBackfillCount(reconnectionManager.estimateBackfill(rate));
            }
            //logger.info("++++++++++++++++++++++++++++++++++++++ ");
            //logger.info("Going to build a request ");
            //logger.info("host : " +host);
            //logger.info("endpointURI: " + endpoint.getURI());
            //logger.info("endpointHTTPMethod : " + endpoint.getHttpMethod());
            //logger.info("auth:" + auth);
            //logger.info("auth:" + auth.toString());
            HttpUriRequest request = HttpConstants.constructRequest(host, endpoint, auth);
            if (request != null) {
                String postContent = null;
                if (endpoint.getHttpMethod().equalsIgnoreCase(HttpConstants.HTTP_POST)) {
                    postContent = endpoint.getPostParamString();
                }
                auth.signRequest(request, postContent);
                Connection conn = new Connection(client, processor);
                StatusLine status = establishConnection(conn, request);
                if (handleConnectionResult(status)) {
                    rateTracker.resume();
                    processConnectionData(conn);
                    rateTracker.pause();
                }
                logger.info("{} Done processing, preparing to close connection", name);
                conn.close();
            } else {
                addEvent(
                        new Event(
                                EventType.CONNECTION_ERROR,
                                String.format("Error creating request: %s, %s, %s", endpoint.getHttpMethod(), host, endpoint.getURI())
                        )
                );
            }
        }
    } catch (Throwable e) {
        logger.warn(name + " Uncaught exception", e);
        Exception laundered = (e instanceof Exception) ? (Exception) e : new RuntimeException(e);
        setExitStatus(new Event(EventType.STOPPED_BY_ERROR, laundered));
    } finally {
        rateTracker.stop();
        logger.info("{} Shutting down httpclient connection manager", name);
        client.getConnectionManager().shutdown();
        isRunning.countDown();
    }
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
/**
 * @return whether a successful connection has been established
 */
@VisibleForTesting
boolean handleConnectionResult(@Nullable StatusLine statusLine) {
    statsReporter.incrNumConnects();
    if (statusLine == null) {
        logger.warn("{} failed to establish connection properly", name);
        addEvent(new Event(EventType.CONNECTION_ERROR, "Failed to establish connection properly"));
        return false;
    }
    int statusCode = statusLine.getStatusCode();
    if (statusCode == HttpConstants.Codes.SUCCESS) {
        logger.debug("{} Connection successfully established", name);
        statsReporter.incrNum200s();
        connectionEstablished.set(true);
        addEvent(new HttpResponseEvent(EventType.CONNECTED, statusLine));
        reconnectionManager.resetCounts();
        return true;
    }

    logger.warn(name + " Error connecting w/ status code - {}, reason - {}", statusCode, statusLine.getReasonPhrase());
    statsReporter.incrNumConnectionFailures();
    addEvent(new HttpResponseEvent(EventType.HTTP_ERROR, statusLine));
    if (HttpConstants.FATAL_CODES.contains(statusCode)) {
        setExitStatus(new Event(EventType.STOPPED_BY_ERROR, "Fatal error code: " + statusCode));
    } else if (statusCode < 500 && statusCode >= 400) {
        statsReporter.incrNum400s();
        // we will retry these a set number of times, then fail
        if (reconnectionManager.shouldReconnectOn400s()) {
            logger.debug("{} Reconnecting on {}", name, statusCode);
            reconnectionManager.handleExponentialBackoff();
        } else {
            logger.debug("{} Reconnecting retries exhausted for {}", name, statusCode);
            setExitStatus(new Event(EventType.STOPPED_BY_ERROR, "Retries exhausted"));
        }
    } else if (statusCode >= 500) {
        statsReporter.incrNum500s();
        reconnectionManager.handleExponentialBackoff();
    } else {
        setExitStatus(new Event(EventType.STOPPED_BY_ERROR, statusLine.getReasonPhrase()));
    }
    return false;
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
private void setExitStatus(Event event) {
    logger.info("{} exit event - {}", name, event.getMessage());
    addEvent(event);
    exitEvent.set(event);
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
public Event getExitEvent() {
    if (!isDone()) {
        throw new IllegalStateException(name + " Still running");
    }
    return exitEvent.get();
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
@Start
public void start() {
    queue = new LinkedBlockingQueue<>(100000);
    BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<>(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();

    // set up some track terms
    if (trackTerms != null && !trackTerms.isEmpty()) {
        hosebirdEndpoint.trackTerms(Lists.newArrayList(trackTerms.split(" ")));
    }
    // set up some followings
    if (followingIDs != null && !followingIDs.isEmpty()) {
        Set<Long> followings = new HashSet<>();
        for (String id: followingIDs.split(" ")) {
            followings.add(Long.parseLong(id));
        }
        hosebirdEndpoint.followings(Lists.newArrayList(followings));
    }

    // These secrets should be read from a config file
    Authentication hosebirdAuth = new OAuth1(consumerKey, consumerSecret, token, secret);

    ClientBuilder builder = new ClientBuilder()
            .name("twitter-client")
            .hosts(hosebirdHosts)
            .authentication(hosebirdAuth)
            .endpoint(hosebirdEndpoint)
            .processor(new StringDelimitedProcessor(queue))
            .eventMessageQueue(eventQueue);

    client = builder.build();
    // Attempts to establish a connection.
    client.connect();

    executor.submit(() -> {
        while (client != null && !client.isDone()) {
            try {
                String msg = queue.poll(5000, TimeUnit.MILLISECONDS);
                if (msg != null) {
                    out.send(msg, null);
                }
            } catch (InterruptedException e) {
                Log.warn("Twitter messages blocking queue interrupted while waiting.");
            }
        }
    });
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
/**
 * This method should only be called after the client is done
 */
public Event getExitEvent() {
  return clientBase.getExitEvent();
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
@Override
public void run() {
  // establish the initial connection
  //   if connection fails due to auth or some other 400, stop immediately
  //   if connection fails due to a 500, back off and retry
  //   if no response or other code, stop immediately
  // begin reading from the stream
  // while the stop signal hasn't been sent, and no IOException from processor, keep processing
  // if  IOException, time to restart the connection:
  //   handle http connection cleanup
  //   do some backoff, set backfill
  // if stop signal set, time to kill/clean the connection, and end this thread.
  try {
    if (client instanceof RestartableHttpClient) {
      ((RestartableHttpClient) client).setup();
    }
    rateTracker.start();
    while (!isDone()) {
      String host = hosts.nextHost();
      if (host == null) {
        setExitStatus(new Event(EventType.STOPPED_BY_ERROR, "No hosts available"));
        break;
      }

      double rate = rateTracker.getCurrentRateSeconds();
      if (!Double.isNaN(rate)) {
        endpoint.setBackfillCount(reconnectionManager.estimateBackfill(rate));
      }

      HttpUriRequest request = HttpConstants.constructRequest(host, endpoint, auth);
      if (request != null) {
        String postContent = null;
        if (endpoint.getHttpMethod().equalsIgnoreCase(HttpConstants.HTTP_POST)) {
          postContent = endpoint.getPostParamString();
        }
        auth.signRequest(request, postContent);
        Connection conn = new Connection(client, processor);
        StatusLine status = establishConnection(conn, request);
        if (handleConnectionResult(status)) {
          rateTracker.resume();
          processConnectionData(conn);
          rateTracker.pause();
        }
        logger.info("{} Done processing, preparing to close connection", name);
        conn.close();
      } else {
        addEvent(
          new Event(
            EventType.CONNECTION_ERROR,
            String.format("Error creating request: %s, %s, %s", endpoint.getHttpMethod(), host, endpoint.getURI())
          )
        );
      }
    }
  } catch (Throwable e) {
    logger.warn(name + " Uncaught exception", e);
    Exception laundered = (e instanceof Exception) ? (Exception) e : new RuntimeException(e);
    setExitStatus(new Event(EventType.STOPPED_BY_ERROR, laundered));
  } finally {
    rateTracker.stop();
    logger.info("{} Shutting down httpclient connection manager", name);
    client.getConnectionManager().shutdown();
    isRunning.countDown();
  }
}
 

import com.twitter.hbc.core.event.Event; //导入依赖的package包/类
/**
 * @return whether a successful connection has been established
 */
@VisibleForTesting
boolean handleConnectionResult(@Nullable StatusLine statusLine) {
  statsReporter.incrNumConnects();
  if (statusLine == null) {
    logger.warn("{} failed to establish connection properly", name);
    addEvent(new Event(EventType.CONNECTION_ERROR, "Failed to establish connection properly"));
    return false;
  }
  int statusCode = statusLine.getStatusCode();
  if (statusCode == HttpConstants.Codes.SUCCESS) {
    logger.debug("{} Connection successfully established", name);
    statsReporter.incrNum200s();
    connectionEstablished.set(true);
    addEvent(new HttpResponseEvent(EventType.CONNECTED, statusLine));
    reconnectionManager.resetCounts();
    return true;
  }

  logger.warn(name + " Error connecting w/ status code - {}, reason - {}", statusCode, statusLine.getReasonPhrase());
  statsReporter.incrNumConnectionFailures();
  addEvent(new HttpResponseEvent(EventType.HTTP_ERROR, statusLine));
  if (HttpConstants.FATAL_CODES.contains(statusCode)) {
    setExitStatus(new Event(EventType.STOPPED_BY_ERROR, "Fatal error code: " + statusCode));
  } else if (statusCode < 500 && statusCode >= 400) {
    statsReporter.incrNum400s();
    // we will retry these a set number of times, then fail
    if (reconnectionManager.shouldReconnectOn400s()) {
      logger.debug("{} Reconnecting on {}", name, statusCode);
      reconnectionManager.handleExponentialBackoff();
    } else {
      logger.debug("{} Reconnecting retries exhausted for {}", name, statusCode);
      setExitStatus(new Event(EventType.STOPPED_BY_ERROR, "Retries exhausted"));
    }
  } else if (statusCode >= 500) {
    statsReporter.incrNum500s();
    reconnectionManager.handleExponentialBackoff();
  } else {
    setExitStatus(new Event(EventType.STOPPED_BY_ERROR, statusLine.getReasonPhrase()));
  }
  return false;
}