FriendOrFollow.scala

/**
 * This program has come from http://gist.github.com/18268 .
 */
import java.net.{Authenticator, PasswordAuthentication}
import java.net.{URL, HttpURLConnection}
import scala.xml.{XML, Node}
import scala.collection.mutable.Map

class FriendOrFollow (screen_name:String) {

  private val friends_url = "http://twitter.com/statuses/friends/%s.xml"
  lazy val friends:Map[String, Node] =
    getMap(friends_url.replace("%s", screen_name))

  private val followers_url = "http://twitter.com/statuses/followers/%s.xml"
  lazy val followers:Map[String, Node] =
    getMap(followers_url.replace("%s", screen_name))

  // intersection of sets of friends and followers users.
  lazy val friendsandfollowers = friends.keySet ** followers.keySet

  lazy val not_following_me_back:Map[String, Node] =
    friends.clone -- friendsandfollowers.elements

  lazy val not_following_them_back:Map[String, Node] =
    followers.clone -- friendsandfollowers.elements

  private final def getMap(url:String):Map[String, Node] = {

    def page(n:Int, map:Map[String, Node]):Map[String, Node] = {
      var urlConn = new URL(url+"?page="+n).openConnection match {
        case con:HttpURLConnection => {
          con.connect();
          con.getResponseCode match {
            case 200 => { }
            case _ => { }
          }
          con
        }
      }

      val size=map.size

      (XML.load(urlConn.getInputStream) \\ "user").foldLeft(map){
        (m, user)=> m + ((user \ "id" text) -> user)
      }
      match {
        case _map if(_map.size<=size) => map
        case _map => page(n+1, _map);
      }
    }

    page(1, Map.empty[String, Node])
  }
}

object FriendOrFollowTest {
  def main(arg:Array[String])={

    val (screen_name, username, passwd) = (arg(0), arg(1), arg(2))

    Authenticator.setDefault(
      new Authenticator {
        override def getPasswordAuthentication = {
          new PasswordAuthentication(username, passwd.toCharArray);
        }
      }
    )

    val fof = new FriendOrFollow(screen_name)

    println(screen_name+" is following these people, "+
            "but they are not following "+screen_name+" back.")

    fof.not_following_me_back.foreach { case (id, node)=>
      println(node \ "screen_name" text)
    }

    println("These people are following "+screen_name+", "+
            "but "+screen_name+" is not following them back.")

    fof.not_following_them_back.foreach { case (id, node)=>
      println(node \ "screen_name" text)
    }

  }
}

TwitterStreamingAPI.scala

import scala.xml.{XML, Elem}

object http{
  import java.net.URLEncoder.encode
  import java.net._
  import java.net.{URL, HttpURLConnection}
  import java.io.{InputStream, IOException}
 
  private def param2str(param:(String,String)*):String =
    (for((k, v)<-param)
       yield k+"="+URLEncoder.encode(v, "UTF-8")).mkString("&")

  private def consume(in:InputStream, f:Option[String]=>Unit){
    val buf = new Array[Byte](1024)
    try{

      for(i <- Stream.const(()=>in.read(buf)).map(_()).takeWhile(_ != -1)){
        f(Some(new String(buf, 0, i)))
      }

      // Scala 2.8 will have Stream#continually method.
      // for(i <- Stream.continually(in.read(buf)).takeWhile(_ != -1)){...}
/*
      def loop() {
        in.read(buf) match{
          case -1 =>
          case i => f(Some(new String(buf, 0, i))); loop()
        }
      }
      loop()
*/

      f(None)
    }
    catch{ case e:IOException => }
    finally{ in.close }
  }

  def get(uri:String, param:(String,String)*)(f: Option[String]=>Unit) = {
    new URL(uri + "?" + param2str(param:_*)).openConnection match{
      case c:HttpURLConnection =>
        c.setRequestMethod("GET")
        consume(c.getInputStream, f)
      case _ =>
    }
  }

  def post(uri:String, param:(String,String)*)(f: Option[String]=>Unit)={
    new URL(uri).openConnection match{
      case c:HttpURLConnection => {
        c.setDoInput(true)
        c.setDoOutput(true)
        c.setUseCaches(false)
        c.setRequestMethod("POST")
        c.setRequestProperty("Content-Type",
                             "application/x-www-form-urlencoded")
        val content = param2str(param:_*).getBytes
        c.setRequestProperty("Content-Length", content.length.toString);
        val o = c.getOutputStream
        o.write(content)
        o.flush
        o.close

        consume(c.getInputStream, f)
      }
      case _ => None
    }
  }
}

object TwitterStreamingAPI{
  import scala.collection.mutable.{Queue, SynchronizedQueue}

  private val follow = "http://stream.twitter.com/follow.xml"
  private val spritzer = "http://stream.twitter.com/spritzer.xml"
  private val track = "http://stream.twitter.com/track.xml"

  // ?s will enable the DOTALL mode
  // *? is the reluctant quantifier, and not greedy.
  private val pattern_limit ="^((?s).*?)</limit>((?s).*)".r
  private val pattern_delete ="^((?s).*?)</delete>((?s).*)".r
  private val pattern_status ="^((?s).*?)</status>((?s).*)".r
  private def parseStatus(queue:Queue[Elem]) = {
    var input = ""
    val proc:Option[String]=>Unit = {
      case Some(_input) =>
        input = (input + _input) match{
          case pattern_limit(_, y) => y
          case pattern_delete(_, y) => y
          case pattern_status(x, y) =>
            queue += XML.loadString(x.trim+"</status>")
            y
          case _input => _input
        }
      case _ => 
    }
    proc
  }

  private def spawnQueueReader(f:(Elem) => Unit):Queue[Elem]={
    val queue = new SynchronizedQueue[Elem]

    import scala.concurrent.ops.spawn
    spawn{
      def loop:Unit = queue.dequeueFirst((_)=>true) match{
        case Some(e) => f(e); loop
        case _ =>
      }

      while(true){
        loop
        Thread.sleep(100)
      }
    }

    queue
  }

  def follow(id:Seq[String])(f:(Elem) => Unit){
    val queue = spawnQueueReader(f)
    val _id = id.take(200).mkString(" ")
    http.post(follow, ("follow", _id))(parseStatus(queue))
  }

  def spritzer(f:(Elem) => Unit){
    val queue = spawnQueueReader(f)
    http.get(spritzer)(parseStatus(queue))
  }

  def track(keyword:Seq[String])(f:(Elem) => Unit){
    val queue = spawnQueueReader(f)
    val _keyword = keyword.take(50).mkString(",")
    http.post(track, ("track", _keyword))(parseStatus(queue))
  }

}

object RT {

  def main(arg:Array[String]){
    import java.net.{Authenticator, PasswordAuthentication}
    val (username, passwd) = (arg(0), arg(1))
    Authenticator.setDefault(
      new Authenticator {
        override def getPasswordAuthentication = {
          new PasswordAuthentication(username, passwd.toCharArray);
        }
      }
    )

    TwitterStreamingAPI.spritzer {
      case e =>
        var (text, user_name) = (e \ "text" text, e \ "user" \ "name" text)
        if(text.indexOf("RT ") != -1)
          println(user_name+": "+text)
    }
  }
}


object UpdatesOfFollowers {

  def main(arg:Array[String]){
    import java.net.{Authenticator, PasswordAuthentication}
    val (username, passwd) = (arg(0), arg(1))
    Authenticator.setDefault(
      new Authenticator {
        override def getPasswordAuthentication = {
          new PasswordAuthentication(username, passwd.toCharArray);
        }
      }
    )

    val followers = new FriendOrFollow(username).followers.keys.collect

    TwitterStreamingAPI.follow(followers) {
      case e =>
        var (text, user_name) = (e \ "text" text, e \ "user" \ "name" text)
        println(user_name+": "+text)
    }
  }
}

object Track {
  def main(arg:Array[String]){
    import java.net.{Authenticator, PasswordAuthentication}
    val Array(username, passwd, rest@_*) = arg
    Authenticator.setDefault(
      new Authenticator {
        override def getPasswordAuthentication = {
          new PasswordAuthentication(username, passwd.toCharArray);
        }
      }
    )

    TwitterStreamingAPI.track(rest) {
      case e =>
        var (text, user_name) = (e \ "text" text, e \ "user" \ "name" text)
        println(user_name+": "+text)
    }
  }
}