slapped together bits of a basic http4s interface to the Twitter streaming API. OAuth not included.

twitter.scala

// a simple process1 that splits the frames up based on a "\r\n" sequence of chars
val partitioner = {
  import java.util.regex.Pattern
  import scalaz.stream.process1
  import scalaz.std.string._

  val pattern = Pattern.compile("\r\n")
  process1.repartition { s: String =>
    pattern.split(s, -1)
  }
}

def makeStream(req: Request): scalaz.stream.Process[Task,Json] =
  await(client.prepare(req).map {
    // pattern match the response code
    case Successful(resp) => 
      (resp.body |> text.utf8Decode |> partitioner) // text is a package of scalaz-stream
        .flatMap { s =>           // before the flatmap we have a Process[Task,String]
          Parse.parse(s) match {  // argonaut json parsing
            case \/-(v) => emit(v)
            case -\/(e) => 
              logger.warn(s"Failure to parse: $e")
              halt
          }
        }
    case resp => sys.error(
      "Failure: " + resp + resp.as[String].run
    )
  })(identity)