package couchdb.changes

import akka.actor.{ActorRef, Actor, ActorLogging, Status}

import akka.io.IO

import scala.concurrent.ExecutionContext.Implicits.global

import scala.language.postfixOps

import scala.language.reflectiveCalls

import scala.util.Try

import spray.can.Http

import spray.can.Http.ConnectionAttemptFailedException

import spray.client.pipelining._

import spray.httpx.unmarshalling.{MalformedContent, Unmarshaller, Deserialized}

import spray.http._

import spray.json._

case class ResultRow(seq: Int, id: String, doc: JsObject)

trait ChangesMarshaller {

  implicit object ChangesMarshaller extends Unmarshaller[ResultRow] {

    import DefaultJsonProtocol._

    implicit val changesFormat = jsonFormat3(ResultRow)

  }

}

class ChangesStreamActor(processor: ActorRef) extends Actor with ActorLogging

  with ChangesMarshaller with settings {

  val io = IO(Http)(context.system)

  def config = context.system.settings.config

  def receive: Receive = {

    case Start =>

      val uri = Uri(s"${couchConf.url}/_changes").withQuery(

          "feed" -> "continuous"

        , "heartbeat" -> couchConf.heartbeat.toMillis.toString

        , "include_docs" -> "true"

      )

      val rq = HttpRequest(HttpMethods.GET, uri = uri)

      log.debug("Sending request {}", rq)

      sendTo(io).withResponsesReceivedBy(self)(rq)

    case x@ChunkedResponseStart(_) =>

      log.debug("Received {}", x)

    case MessageChunk(data, _) =>

      log.debug("Received data chunk {}", data)

      if (data.length > 2) // skip heartbeat messages

        ChangesMarshaller(data).fold(

          _ => log.warning("Failed to deserialize entity {}", data),

          processor !

        )

    case Status.Failure(t) =>

      log.error("Received failure {}", t)

      t match {

        case x:ConnectionAttemptFailedException =>

          log.info("Scheduling reconnect to couchdb in {}", couchConf.reconnect)

          context.system.scheduler.scheduleOnce(couchConf.reconnect, self,

            Start)

        case _ =>

      }

    case x =>

      log.debug("Received {}", x)

  }

}

// vim: set ts=2 sw=2 et: