scalaz-stream + jdbc

scalaz-stream-jdbc.scala

import java.sql.DriverManager
import scalaz.concurrent.Task
import scalaz.stream.io
import scalaz.stream.Process.End

class Query(query: String, chunkSize: Int) { 
  Class forName "org.postgresql.Driver"
  val connection = DriverManager.getConnection(url, "username", "password")
  val statement = connection.createStatement
  // This will cause the postgres driver to create a cursor and give ResultSets of that size
  statement.setFetchSize(chunkSize)
  val resultSet = statement.executeQuery(query)
  def next = resultSet.next
  def isAfterLast = resultSet.isAfterLast
  def getString(i:Int) = resultSet.getString(i)
  def close { 
    resultSet.close
    statement.close
    connection.close 
  } 
}

val chunkSize = 100

val query: scalaz.stream.Process[Task, IndexedSeq[String]] = io.resource(
  Task.delay(new Query("select * from foo", chunkSize))
)(rs =>
  Task.delay(rs.close)
)(rs =>
  Task.delay{
    if (rs.isAfterLast) throw End
    for (_ <- 1 to chunkSize if rs.next) yield rs.getString(1)
  }
)