<commit_before>package com.linkedin.led.twitter.streaming

import java.io.IOException
import org.apache.commons.httpclient.DefaultHttpMethodRetryHandler
import org.apache.commons.httpclient.HttpClient
import org.apache.commons.httpclient.HttpException
import org.apache.commons.httpclient.HttpStatus
import org.apache.commons.httpclient.HttpURL
import org.apache.commons.httpclient.UsernamePasswordCredentials
import org.apache.commons.httpclient.auth.AuthScope
import org.apache.commons.httpclient.HttpMethod
import org.apache.commons.httpclient.params.HttpMethodParams

/**
 * Twitter Streaming API client
 *
 * @param username the twitter account username
 * @param password the twitter account password
 * @param streamProcessor handles the processing of the stream
 *
 * @author <a href="mailto:<EMAIL>"><NAME></a>
 */
class StreamingClient(val username: String, val password: String, streamProcessor: StreamProcessor) extends StreamingMethods {

  /**
   * Back off strategy is:
   * TCP errors start at 250 miliseconds and cap at 16 seconds
   * HTTP errors start at 10 seconds and cap at 240 seconds
   */
  val tcpBackOff  = BackOff(250, 16000)
  val httpBackOff = BackOff(10000, 240000)

  /**
   * stream is a recursive method that fetches the stream
   * in the case of errors, backs off.
   *  @param method the http method to use with the necessary parameters already set
   */
  def stream(method: HttpMethod): Unit = {
    val httpClient: HttpClient = this.getStreamingHttpClient(method)

    // Use compression and a custom user agent
    method.setRequestHeader("Accept-Encoding", "gzip")
    method.setRequestHeader("User-Agent", Config.readString("userAgent"))

    try {
      httpClient.executeMethod(method)

      // Check we received the correct HTTP status on the first attempt to connect
      if (method.getStatusCode() != HttpStatus.SC_OK) {
        throw new HttpException("There was a problem connecting, HTTP code received was: " + 
        method.getStatusCode() +" "+ method.getStatusLine())
      }

      // Reset the errors since the request was successful
      this.resetBackOffs()

      // Let's delegate the processing to the StreamProcessor object
      // You must override the process method in order to do a custom processing with the stream
      streamProcessor.process(method.getResponseBodyAsStream())
    } catch {
      case e: InterruptedException => return;
      case e: HttpException => httpBackOff.backOff
      case e: IOException   => tcpBackOff.backOff
      case _ =>
    } finally {
      method.abort
      method.releaseConnection
    }

    // This is a recursive method that internally backs off when there's an exception or error processing
    stream(method)
  }

  /**
   * Returns an HttpClient object adding the necessary configuration params.
   */
  def getStreamingHttpClient(method: HttpMethod): HttpClient = {
    val httpClient = new HttpClient
    // Sets the default socket timeout (SO_TIMEOUT) in milliseconds which is the timeout for waiting for data.
    // A timeout value of zero is interpreted as an infinite timeout.
    httpClient.getHttpConnectionManager.getParams.setSoTimeout(Config.readInt("socketTimeout"))
    // Default recovery procedure can be replaced with a custom one, let's do that.<commit_msg>Make sure the stream call is optimized as recursive<commit_after>package com.linkedin.led.twitter.streaming

import java.io.IOException
import org.apache.commons.httpclient.DefaultHttpMethodRetryHandler
import org.apache.commons.httpclient.HttpClient
import org.apache.commons.httpclient.HttpException
import org.apache.commons.httpclient.HttpStatus
import org.apache.commons.httpclient.HttpURL
import org.apache.commons.httpclient.UsernamePasswordCredentials
import org.apache.commons.httpclient.auth.AuthScope
import org.apache.commons.httpclient.HttpMethod
import org.apache.commons.httpclient.params.HttpMethodParams

/**
 * Twitter Streaming API client
 *
 * @param username the twitter account username
 * @param password the twitter account password
 * @param streamProcessor handles the processing of the stream
 *
 * @author <a href="mailto:<EMAIL>"><NAME></a>
 */
class StreamingClient(val username: String, val password: String, streamProcessor: StreamProcessor) extends StreamingMethods {

  /**
   * Back off strategy is:
   * TCP errors start at 250 miliseconds and cap at 16 seconds
   * HTTP errors start at 10 seconds and cap at 240 seconds
   */
  val tcpBackOff  = BackOff(250, 16000)
  val httpBackOff = BackOff(10000, 240000)

  /**
   * stream is a recursive method that fetches the stream
   * in the case of errors, backs off.
   *  @param method the http method to use with the necessary parameters already set
   */
  final def stream(method: HttpMethod): Unit = {
    try {
      val httpClient: HttpClient = this.getStreamingHttpClient(method)
      
      // Use compression and a custom user agent
      method.setRequestHeader("Accept-Encoding", "gzip")
      method.setRequestHeader("User-Agent", Config.readString("userAgent"))
      
      try {
        httpClient.executeMethod(method)
      
        // Check we received the correct HTTP status on the first attempt to connect
        if (method.getStatusCode() != HttpStatus.SC_OK) {
          throw new HttpException("There was a problem connecting, HTTP code received was: " + 
          method.getStatusCode() +" "+ method.getStatusLine())
        }
      
        // Reset the errors since the request was successful
        this.resetBackOffs()
      
        // Let's delegate the processing to the StreamProcessor object
        // You must override the process method in order to do a custom processing with the stream
        streamProcessor.process(method.getResponseBodyAsStream())
      } catch {
        case e: InterruptedException => return;
        case e: HttpException => httpBackOff.backOff
        case e: IOException   => tcpBackOff.backOff
        case _ =>
      }      
    } finally {
      method.abort
      method.releaseConnection
    }

    // This is a recursive method that internally backs off when there's an exception or error processing
    stream(method)
  }

  /**
   * Returns an HttpClient object adding the necessary configuration params.
   */
  def getStreamingHttpClient(method: HttpMethod): HttpClient = {
    val httpClient = new HttpClient
    // Sets the default socket timeout (SO_TIMEOUT) in milliseconds which is the timeout for waiting for data.
    // A timeout value of zero is interpreted as an infinite timeout.
    httpClient.getHttpConnectionManager.getParams.setSoTimeout(Config.readInt("socketTimeout"))
    // Default recovery procedure can be replaced with a custom one, let's do that.
{'fraction_non_alphanumeric': 0.07728894173602854, 'fraction_numerical': 0.008620689655172414, 'mean_word_length': 3.5374241402562374, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2145102', 'n_tokens_mistral': 1770, 'n_tokens_neox': 1696, 'n_words': 607}