Simple Connection Level Akka Http Example

ConnectionLevel.scala

package httpclient

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.http.scaladsl.model.StatusCodes._
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._
import spray.json._

import scala.concurrent.{Await, Future}
import scala.concurrent.duration._

/**
  * Created by ASRagab on 2/23/16.
  */


object JsonProtocol extends SprayJsonSupport with DefaultJsonProtocol {
  case class IpInfo(ip: String)
  implicit val format = jsonFormat1(IpInfo.apply)
}

object ConnectionLevel extends App {
  implicit val system = ActorSystem()
  implicit val ec = system.dispatcher
  implicit val materializer = ActorMaterializer()

  val connectionFlow: Flow[HttpRequest, HttpResponse, Future[Http.OutgoingConnection]] = Http().outgoingConnection("api.ipify.org")
  val responseFuture =
    Source.single(HttpRequest(uri = "/?format=json"))
      .via(connectionFlow)
      .runWith(Sink.headOption)

  import JsonProtocol._

  responseFuture map { res =>
    res.get.status match {
      case OK =>
        Unmarshal(res.get.entity).to[IpInfo].map { info =>
          println(s"The Information for my ip is $info")
          shutdown
        }
      case _ =>  Unmarshal(res.get.entity).to[String].map { cause =>
        println(s"Response status ${res.get.entity} received because $cause")
        shutdown
      }

    }
  }

  def shutdown() = {
    system.terminate
    Await.result(system.whenTerminated, 5 seconds)
  }
}