<commit_before>/*
 * Copyright 2016 <NAME>, <EMAIL>
 *
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.amient.affinity.core.actor

import java.security.cert.X509Certificate
import java.security.{KeyStore, SecureRandom}
import java.util
import java.util.concurrent.ConcurrentHashMap
import javax.net.ssl.{KeyManagerFactory, SSLContext, X509TrustManager}

import akka.actor.{Actor, ActorRef, PoisonPill, Props, Terminated}
import akka.event.Logging
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model.StatusCodes._
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.ws.{BinaryMessage, Message, TextMessage, UpgradeToWebSocket}
import akka.pattern.ask
import akka.routing._
import akka.serialization.SerializationExtension
import akka.stream.ActorMaterializer
import akka.stream.actor.ActorPublisher
import akka.stream.actor.ActorPublisherMessage.Request
import akka.stream.scaladsl.{Flow, Sink, Source}
import akka.util.{ByteString, Timeout}
import com.fasterxml.jackson.databind.JsonNode
import io.amient.affinity.core.ack
import io.amient.affinity.core.actor.Cluster.{CheckClusterAvailability, ClusterAvailability}
import io.amient.affinity.core.actor.Controller.GracefulShutdown
import io.amient.affinity.core.cluster.Coordinator.MasterStatusUpdate
import io.amient.affinity.core.http.RequestMatchers.{HTTP, PATH}
import io.amient.affinity.core.http.{Encoder, HttpExchange, HttpInterface}
import io.amient.affinity.core.serde.avro.{AvroRecord, AvroSerde}
import io.amient.affinity.core.util.{ByteUtils, Reply}
import org.apache.avro.util.ByteBufferInputStream

import scala.collection.JavaConverters._
import scala.concurrent.duration._
import scala.concurrent.{Await, ExecutionContext, Future, Promise}
import scala.util.control.NonFatal
import scala.language.postfixOps
import scala.reflect.ClassTag

object Gateway {
  final val CONFIG_HTTP_HOST = "affinity.node.gateway.http.host"
  final val CONFIG_HTTP_PORT = "affinity.node.gateway.http.port"

  final val CONFIG_TLS_KEYSTORE_PKCS = "affinity.node.gateway.tls.keystore.standard"
  final val CONFIG_TLS_KEYSTORE_PASSWORD = "affinity.node.gateway.tls.keystore.password"
  final val CONFIG_TLS_KEYSTORE_RESOURCE = "affinity.node.gateway.tls.keystore.resource"
  final val CONFIG_TLS_KEYSTORE_FILE = "affinity.node.gateway.tls.keystore.file"
}

abstract class Gateway extends Actor {

  import Gateway._

  private val log = Logging.getLogger(context.system, this)

  private val config = context.system.settings.config

  private val services = new ConcurrentHashMap[Class[_ <: Actor], ActorRef]

  val cluster = context.actorOf(Props(new Cluster()), name = "cluster")

  import context.system
  import context.dispatcher
  private implicit val scheduler = context.system.scheduler

  val sslContext = if (!config.hasPath(CONFIG_TLS_KEYSTORE_PASSWORD)) None else Some(SSLContext.getInstance("TLS"))
  sslContext.foreach { context =>
    log.info("Configuring SSL Context")
    val password = config.getString(CONFIG_TLS_KEYSTORE_PASSWORD).toCharArray
    val ks = KeyStore.getInstance(config.getString(CONFIG_TLS_KEYSTORE_PKCS))
    if (config.hasPath(CONFIG_TLS_KEYSTORE_RESOURCE)) {
      ks.load(getClass.getClassLoader.getResourceAsStream(config.getString(CONFIG_TLS_KEYSTORE_RESOURCE)), password)
    } else {
      ks.load(getClass.getClassLoader.getResourceAsStream(config.getString(CONFIG_TLS_KEYSTORE_FILE)), password)
    }
    val keyManagerFactory = KeyManagerFactory.getInstance("SunX509")
    keyManagerFactory.init(ks, password)
    context.init(keyManagerFactory.getKeyManagers, null, new SecureRandom)
  }

  private val httpInterface: HttpInterface = new HttpInterface(
    config.getString(CONFIG_HTTP_HOST), config.getInt(CONFIG_HTTP_PORT), sslContext)


  def describeServices = services.asScala.map { case (k, v) => (k.toString, v.path.toString) }

  def describeRegions: List[String] = {
    val t = 60 seconds
    implicit val timeout = Timeout(t)
    Await.result(cluster ? GetRoutees, t).asInstanceOf[Routees].routees.map(_.toString).toList.sorted
  }

  override def preStart(): Unit = {
    log.info("starting gateway")
    httpInterface.bind(self)
    context.watch(cluster)
    context.parent ! Controller.GatewayCreated(httpInterface.getListenPort)
  }

  override def postStop(): Unit = {
    log.info("stopping gateway")
    httpInterface.close()
  }

  def service[X <: Actor](implicit tag: ClassTag[X]): ActorRef = {
    services.get(tag.runtimeClass) match {
      case null => throw new IllegalStateException(s"Service not available for ${tag.runtimeClass}")
      case instance => instance
    }
  }


  def handleException: PartialFunction[Throwable, HttpResponse] = {
    case e: NoSuchElementException => HttpResponse(NotFound)
    case e: IllegalArgumentException => HttpResponse(BadRequest)
    case e: UnsupportedOperationException => HttpResponse(NotImplemented)
    case e: IllegalStateException => HttpResponse(ServiceUnavailable)
    case NonFatal(e) => e.printStackTrace(); HttpResponse(InternalServerError)
  }

  def fulfillAndHandleErrors(promise: Promise[HttpResponse])(f: => Future[HttpResponse])
                            (implicit ctx: ExecutionContext) {
    promise.completeWith(f recover handleException)
  }

  def delegateAndHandleErrors(promise: Promise[HttpResponse], delegate: Future[Any])
                             (f: Any => HttpResponse)(implicit ctx: ExecutionContext) {
    promise.completeWith(delegate map f recover handleException)
  }

  def delegateAsJson(response: Promise[HttpResponse], delegate: Reply[_])(implicit timeout: Timeout): Unit = {
    delegateAsJson(response, cluster, delegate)
  }<commit_msg>for standard runtime exceptions http plain text entities are send as response alongside the appropriate error status code<commit_after>/*
 * Copyright 2016 <NAME>, <EMAIL>
 *
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.amient.affinity.core.actor

import java.security.cert.X509Certificate
import java.security.{KeyStore, SecureRandom}
import java.util
import java.util.concurrent.ConcurrentHashMap
import javax.net.ssl.{KeyManagerFactory, SSLContext, X509TrustManager}

import akka.actor.{Actor, ActorRef, PoisonPill, Props, Terminated}
import akka.event.Logging
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model.StatusCodes._
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.ws.{BinaryMessage, Message, TextMessage, UpgradeToWebSocket}
import akka.pattern.ask
import akka.routing._
import akka.serialization.SerializationExtension
import akka.stream.ActorMaterializer
import akka.stream.actor.ActorPublisher
import akka.stream.actor.ActorPublisherMessage.Request
import akka.stream.scaladsl.{Flow, Sink, Source}
import akka.util.{ByteString, Timeout}
import com.fasterxml.jackson.databind.JsonNode
import io.amient.affinity.core.ack
import io.amient.affinity.core.actor.Cluster.{CheckClusterAvailability, ClusterAvailability}
import io.amient.affinity.core.actor.Controller.GracefulShutdown
import io.amient.affinity.core.cluster.Coordinator.MasterStatusUpdate
import io.amient.affinity.core.http.RequestMatchers.{HTTP, PATH}
import io.amient.affinity.core.http.{Encoder, HttpExchange, HttpInterface}
import io.amient.affinity.core.serde.avro.{AvroRecord, AvroSerde}
import io.amient.affinity.core.util.{ByteUtils, Reply}
import org.apache.avro.util.ByteBufferInputStream

import scala.collection.JavaConverters._
import scala.concurrent.duration._
import scala.concurrent.{Await, ExecutionContext, Future, Promise}
import scala.util.control.NonFatal
import scala.language.postfixOps
import scala.reflect.ClassTag

object Gateway {
  final val CONFIG_HTTP_HOST = "affinity.node.gateway.http.host"
  final val CONFIG_HTTP_PORT = "affinity.node.gateway.http.port"

  final val CONFIG_TLS_KEYSTORE_PKCS = "affinity.node.gateway.tls.keystore.standard"
  final val CONFIG_TLS_KEYSTORE_PASSWORD = "affinity.node.gateway.tls.keystore.password"
  final val CONFIG_TLS_KEYSTORE_RESOURCE = "affinity.node.gateway.tls.keystore.resource"
  final val CONFIG_TLS_KEYSTORE_FILE = "affinity.node.gateway.tls.keystore.file"
}

abstract class Gateway extends Actor {

  import Gateway._

  private val log = Logging.getLogger(context.system, this)

  private val config = context.system.settings.config

  private val services = new ConcurrentHashMap[Class[_ <: Actor], ActorRef]

  val cluster = context.actorOf(Props(new Cluster()), name = "cluster")

  import context.system
  import context.dispatcher
  private implicit val scheduler = context.system.scheduler

  val sslContext = if (!config.hasPath(CONFIG_TLS_KEYSTORE_PASSWORD)) None else Some(SSLContext.getInstance("TLS"))
  sslContext.foreach { context =>
    log.info("Configuring SSL Context")
    val password = config.getString(CONFIG_TLS_KEYSTORE_PASSWORD).toCharArray
    val ks = KeyStore.getInstance(config.getString(CONFIG_TLS_KEYSTORE_PKCS))
    if (config.hasPath(CONFIG_TLS_KEYSTORE_RESOURCE)) {
      ks.load(getClass.getClassLoader.getResourceAsStream(config.getString(CONFIG_TLS_KEYSTORE_RESOURCE)), password)
    } else {
      ks.load(getClass.getClassLoader.getResourceAsStream(config.getString(CONFIG_TLS_KEYSTORE_FILE)), password)
    }
    val keyManagerFactory = KeyManagerFactory.getInstance("SunX509")
    keyManagerFactory.init(ks, password)
    context.init(keyManagerFactory.getKeyManagers, null, new SecureRandom)
  }

  private val httpInterface: HttpInterface = new HttpInterface(
    config.getString(CONFIG_HTTP_HOST), config.getInt(CONFIG_HTTP_PORT), sslContext)


  def describeServices = services.asScala.map { case (k, v) => (k.toString, v.path.toString) }

  def describeRegions: List[String] = {
    val t = 60 seconds
    implicit val timeout = Timeout(t)
    Await.result(cluster ? GetRoutees, t).asInstanceOf[Routees].routees.map(_.toString).toList.sorted
  }

  override def preStart(): Unit = {
    log.info("starting gateway")
    httpInterface.bind(self)
    context.watch(cluster)
    context.parent ! Controller.GatewayCreated(httpInterface.getListenPort)
  }

  override def postStop(): Unit = {
    log.info("stopping gateway")
    httpInterface.close()
  }

  def service[X <: Actor](implicit tag: ClassTag[X]): ActorRef = {
    services.get(tag.runtimeClass) match {
      case null => throw new IllegalStateException(s"Service not available for ${tag.runtimeClass}")
      case instance => instance
    }
  }


  def handleException: PartialFunction[Throwable, HttpResponse] = {
    case e: NoSuchElementException => HttpResponse(NotFound, entity = if (e.getMessage == null) "" else e.getMessage)
    case e: IllegalArgumentException => HttpResponse(BadRequest, entity = if (e.getMessage == null) "" else e.getMessage)
    case e: UnsupportedOperationException => HttpResponse(NotImplemented, entity = if (e.getMessage == null) "" else e.getMessage)
    case e: IllegalStateException => HttpResponse(ServiceUnavailable)
    case NonFatal(e) => e.printStackTrace(); HttpResponse(InternalServerError)
  }

  def fulfillAndHandleErrors(promise: Promise[HttpResponse])(f: => Future[HttpResponse])
                            (implicit ctx: ExecutionContext) {
    promise.completeWith(f recover handleException)
  }

  def delegateAndHandleErrors(promise: Promise[HttpResponse], delegate: Future[Any])
                             (f: Any => HttpResponse)(implicit ctx: ExecutionContext) {
    promise.completeWith(delegate map f recover handleException)
  }

  def delegateAsJson(response: Promise[HttpResponse], delegate: Reply[_])(implicit timeout: Timeout): Unit = {
    delegateAsJson(response, cluster, delegate)
  }
{'fraction_non_alphanumeric': 0.09581656669956723, 'fraction_numerical': 0.002885126414091565, 'mean_word_length': 5.0228623685413805, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1563763', 'n_tokens_mistral': 3904, 'n_tokens_neox': 3762, 'n_words': 909}