play 2.6에 엄청 많은 기능이 추가/변경되었다. playframwork는 조금 불친절하다.

https://www.playframework.com/documentation/2.6.x/Highlights26

자세한 내용은 아래 블로그에 play 2.6에 대한 설명이 잘 되어 있다. 

https://nvisium.com/resources/blog/2017/10/04/play-2-6-security-analysis.html

자바 개발자가 https://www.playframework.com/documentation/2.6.x/ScalaHttpFilters를 보면서 스칼라 Play 애플리케이션을 만들 때 조금 헤맬 수 있다. 

아래와 같은 코드가 있다고 가정하자. 특별히 소스를 분석하지 않아도 적당히 문서를 읽으면서 알수도 있지만,,

head first로 개념을 이해할 수 있다. 

Filters

package filters

import javax.inject.Inject

import play.api.http.{DefaultHttpFilters, EnabledFilters}
import play.filters.gzip.GzipFilter

class Filters @Inject() (
                          defaultFilters: EnabledFilters,
                          gzip: GzipFilter,
                          logging: LoggingFilter
                        ) extends DefaultHttpFilters
                          (defaultFilters.filters :+ gzip :+ logging: _*)

LoggingFilter

package filters

import javax.inject.Inject

import akka.stream.Materializer
import play.api.Logger
import play.api.mvc._

import scala.concurrent.{ExecutionContext, Future}

class LoggingFilter @Inject() (implicit val mat: Materializer, 
            ec: ExecutionContext) extends Filter {

  def apply(nextFilter: RequestHeader => Future[Result])
           (requestHeader: RequestHeader): Future[Result] = {

    val startTime = System.currentTimeMillis

    nextFilter(requestHeader).map { result =>

      val endTime = System.currentTimeMillis
      val requestTime = endTime - startTime

      val log = s"${requestHeader.method} ${requestHeader.uri} 
              took ${requestTime}ms and returned ${result.header.status}"
      if (requestTime > 3000) {
        Logger.warn(log)
      } else {
        Logger.debug(log)
      }

      result
    }
  }
}

또는 아래와 같이 사용한다.

play.filters.enabled += filters.LoggingFilter

만약 play.filters.enabled를 사용하지 않으면.. 아래와 같이 써야 한다. 

play.http.filters = filters.Filters

아래와 같이 사용하면  다음 에러가 발생한다. 

play.http.filters += filters.LoggingFilter

play.filters.enabled += filters.Filters

클래스를 보면 Filters의 첫 번째 매개변수인 EnabledFilters의 내부를 보면.. 필터를 구성하는 개념이다. 

즉 Filters는 상위 개념인데, 처음에는 None으로 지정되어 있다.  따러서 ConfigNull이 이미 들어가 있다. 따라서 Null에 List를 추가하면 당연히 에러가 발생할 것이다.  play.http.filters는 애플리케이션에서 지정하는 filter 목록을 정의하는 것이라 할 수 있겠다. 

play.http.filters += filters.Filters

play.http.filters = filters.MyFilters

extends DefaultHttpFilters(defaultFilters.filters :+ gzip :+ logging: _*)

class DefaultHttpFilters @Inject() (val filters: EssentialFilter*)

defaultFilters.filters :+ gzip :+ logging)

 이를 : _*) 를 추가하면  EssentailFilter* 타입이 된다.

자바에서는 varargs인데.

스칼라에서는 : _*으로 사용하면 컴파일러에게 seq/array를 varargs로 변환하라는 신호이다.

https://www.playframework.com/documentation/2.6.x/ScalaWS를 살펴보면 WSClient를 

Injection해서 사용하고 있다.

WSClient를 Injection없이 바로 써서 테스트하고 싶다면 다음 예제를 활용한다. 

import play.api.libs.ws.ahc.AhcWSClient
import akka.stream.ActorMaterializer
import akka.actor.ActorSystem
import scala.concurrent.duration._
import scala.concurrent.{Await, Future}

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
val ws = AhcWSClient()

val result = ws.url("https://www.google.com")
  .withRequestTimeout(10000.millis)
  .get()
  .map{
  resp => resp.body
}(system.dispatcher)

val response = Await.result(result, atMost = 5.second)

println(response)

외부 클래스에서 trait의 내부 필드에 접근할 때는 java interface처럼 접근할 수 없다. 

즉 public처럼 사용할 수 없다.

따라서 컴패년 오브젝트를 사용하면 외부 클래스에서 사용할 수 있다.

scala 프로젝트에서 사용할만한 scala- elasticsearch 라이브러리는 sksamuel이 가장 좋은 것 같다. (아직 갈길이 멀지만.)

sksamuel의 HTTPClient을 잘 사용하고 있다. 

val result = client.execute {
    val req = search(indexName).
      query {
        bool {
         //..
    }

    queryStatement = client.show(req)
    req
  }
}

중요한 것은 비동기로 되어 있어서 쭉 지나가 버린다. 

동기(synchronous) 코드로 개발하려면 Await.result 또는 Await.ready를 사용한다. 

Await.result를 사용하는 코드는 간단하다. 내부적으로 Future[Either[]] 형태로 Right/Left를 먼저 써야 한다. 

import scala.concurrent.duration._
val response = Await.result(result, atMost = 5.second)

response match {
  case Right(success) => {
    // ..

  }
  case Left(e) => {
    //..
  }
}

Await.ready와 Await.result는 exception 처리방식이 조금 다르다. 나는 아직 완벽한 프로그램이 아니라서 차라리 crash가 나은 듯해서 Await.result를 사용해봤다.

exception이 발생하면 Await.result는 crash exception이 발생하고, 

Await.ready는 exception을 Failure로 감싼다. 

scala retry 참조 코드 

https://stackoverflow.com/questions/7930814/whats-the-scala-way-to-implement-a-retry-able-call-like-this-one

import util._

object RetryUtils {
  
  @annotation.tailrec
  def retry[T](n: Int)(fn: => T): T = {
    Try { fn } match {
      case Success(x) => x
      case _ if n > 1 => retry(n - 1)(fn)
      case Failure(e) => throw e
    }
  }
}

예제 코드

val retryResult = retry(3) {
  if (validateObject(esClient)) {
    esClient
  } else {
    null
  }
}

Spark와 Kafka 연동하는 방식은 다음과 같다.

- 수신기 기반 접근 방식(Receiver-based approach)

- 다이렉트 스트림 접근 방식(Direct stream approach)

- 구조화된 스트리밍(Structured streaming)

1. 수신기 기반 접근 방식

수신기 기반 방식은 스파크와 카프카와의 첫 번째 통합 방식이었다. 수신기 접근 방식에서 드라이버는 익스큐터에서 카프카 브로커의 고급 API를 사용해 데이터를 가져올 수 있는 수신자를 실행한다. 수신자가 카프카 브로커에서 이벤트를 가져 오고 있기 때문에 수신자는 주키퍼(zookeeper)에 오프셋을 저장한다. 주키퍼는 카프카 클러스터에서도 사용된다. 주요 측면은 WAL(Write Ahead Log)의 사용이다. 수신자는 카프카에서 데이터를 소비하면서 WAL에 계속 저장한다. 따라서 문제가 발생해 익스큐터 또는 수신자가 손실되거나 재시작될 때 WAL을 사용해 이벤트를 복구하고 처리할 수 ​​있다. 따라서이 로그 기반 설계는 내구성과 일관성을 모두 제공한다.

스트림을 받는 예제는 다음과 같다. 

지연을 처리하기 위해 watermark를 사용할 수 있다. 다음은 그 예이다. 

배열에서 맨 앞 또는 뒤의 엘리먼트를 제거하고 싶을 때가 있다. slice를 사용하면 좋다.

배열의 마지막 엘리먼트를 제거하고 싶다면 다음과 같다. 

또는 dropRight 또는 drop을 사용할 수 있다. 

주의할 점은 0을 사용하면 의미 없다.

전역 변수는 아래와 같이 _를 사용해서 정의할 수 있다.

var i: Int = _

var s: String = _

로컬 변수는 아래와 같이 

var response: LogResponse = _

Play2-React를 개발하던 중에 아래와 같은 에러가 크롬에서 발생했다. 

https://www.playframework.com/documentation/2.6.x/CorsFilter를 참고로 설정을 수정했다.

play.filters.enabled += "play.filters.cors.CORSFilter"

play.filters.cors {

  pathPrefixes = ["/", ...]

  allowedOrigins = ["http://localhost", ...]

  allowedHttpMethods = ["GET", "POST"]

  allowedHttpHeaders = ["Accept"]

  preflightMaxAge = 3 days

}

처음에는 404, 정책을 잘 지정하지 못하면 403이 나타난다.

allowedOrigins = ["http://localhost:9000", ...]

다음과 같이 모두 개방하면 테스트는 편해진다. 상용에서는 보안을 위해서 잘 수정하는 것이 좋다. 

(테스트용)

play.filters.hosts {

  allowed = ["."]

}

(상용)

play.filters.hosts {

    allowed = ["localhost", ".kakao.com", "localhost:9000", "local.google.com:9000", "local.google.com:4200"]

}

클라이언트(js)에서도 다음과 같이 수정한다. 

이제 요청하면 정상적으로 동작하는 지 확인할 수 있다.