<commit_before>package org.squbs.httpclient.pipeline

import akka.actor._
import akka.pattern._
import org.squbs.httpclient.endpoint.Endpoint
import org.squbs.httpclient.pipeline.HttpClientContextUtils._
import org.squbs.pipeline.{PipelineProcessorActor, RequestContext}
import org.squbs.proxy.{PipelineResolverRegistry, SimplePipelineConfig, SimpleProcessor}
import spray.client.pipelining.SendReceive
import spray.http._

/**
 * Created by jiamzhang on 2015/3/6.
 */
class HttpClientPipelineActor(clientName: String, endpoint: Endpoint, pipelineConf: SimplePipelineConfig, target: SendReceive) extends Actor with ActorLogging {

  val pipeplineResolverRegistry = PipelineResolverRegistry(context.system)

	override def receive = {
		case request: HttpRequest =>
      val responder = sender()
      val targetAgent = context.actorOf(Props(classOf[HttpClientPipelineTargetActor], target))
      pipeplineResolverRegistry.default.resolve(pipelineConf) match {
        case None => targetAgent tell(request, responder)
        case Some(proc) =>
          val pipeproxy = context.actorOf(Props(classOf[PipelineProcessorActor], targetAgent, responder, proc))
          context.watch(pipeproxy)
          pipeproxy ! RequestContext(request) +> (("HttpClient.name" -> clientName),("HttpClient.Endpoint" -> endpoint))
      }


		case request: ChunkedRequestStart =>
			// not supported yet
			sender ! HttpResponse(StatusCodes.InternalServerError, HttpEntity("Chunked request is not supported yet"))
//			val responder = sender()
//			val targetAgent = context.actorOf(Props(classOf[HttpClientPipelineTargetActor], target))
//			val pipeproxy = context.actorOf(Props(classOf[PipelineProcessorActor], targetAgent, responder, SimpleProcessor(pipelineConf)))
//			context.watch(pipeproxy)
//			pipeproxy ! RequestContext(request.request, true) +> endpoint

		case Terminated(actor) =>
			log.debug("The actor " + actor + " is terminated, stop myself")
			context.stop(self)
	}
}

class HttpClientPipelineTargetActor(target: SendReceive) extends Actor with ActorLogging {
	import context._
	private var client: ActorRef = ActorRef.noSender

	override def receive = {
		case request: HttpRequest =>
			client = sender()
			target(request) pipeTo self

		case response: HttpResponse =>
			client ! response
	}
}
<commit_msg>enable client specific processor resolver<commit_after>package org.squbs.httpclient.pipeline

import akka.actor._
import akka.pattern.pipe
import org.squbs.httpclient.endpoint.Endpoint
import org.squbs.pipeline.{PipelineProcessorActor, RequestContext}
import org.squbs.proxy.{PipelineResolverRegistry, SimplePipelineConfig, SimpleProcessor}
import spray.client.pipelining.SendReceive
import spray.http._

/**
 * Created by jiamzhang on 2015/3/6.
 */
class HttpClientPipelineActor(clientName: String, endpoint: Endpoint, pipelineConf: SimplePipelineConfig, target: SendReceive) extends Actor with ActorLogging {

  val pipeplineResolverRegistry = PipelineResolverRegistry(context.system)

  override def receive = {
    case request: HttpRequest =>
      val responder = sender()
      val targetAgent = context.actorOf(Props(classOf[HttpClientPipelineTargetActor], target))
      pipeplineResolverRegistry.getResolver(clientName).getOrElse(pipeplineResolverRegistry.default).resolve(pipelineConf) match {
        case None => targetAgent tell(request, responder)
        case Some(proc) =>
          val pipeproxy = context.actorOf(Props(classOf[PipelineProcessorActor], targetAgent, responder, proc))
          context.watch(pipeproxy)
          pipeproxy ! RequestContext(request) +>(("HttpClient.name" -> clientName), ("HttpClient.Endpoint" -> endpoint))
      }


    case request: ChunkedRequestStart =>
      // not supported yet
      sender ! HttpResponse(StatusCodes.InternalServerError, HttpEntity("Chunked request is not supported yet"))
    //			val responder = sender()
    //			val targetAgent = context.actorOf(Props(classOf[HttpClientPipelineTargetActor], target))
    //			val pipeproxy = context.actorOf(Props(classOf[PipelineProcessorActor], targetAgent, responder, SimpleProcessor(pipelineConf)))
    //			context.watch(pipeproxy)
    //			pipeproxy ! RequestContext(request.request, true) +> endpoint

    case Terminated(actor) =>
      log.debug("The actor " + actor + " is terminated, stop myself")
      context.stop(self)
  }
}

class HttpClientPipelineTargetActor(target: SendReceive) extends Actor with ActorLogging {
  import context._
  private var client: ActorRef = ActorRef.noSender

  override def receive = {
    case request: HttpRequest =>
      client = sender()
      target(request) pipeTo self

    case response: HttpResponse =>
      client ! response
  }
}

{'fraction_non_alphanumeric': 0.0977730646871686, 'fraction_numerical': 0.002545068928950159, 'mean_word_length': 4.620977353992848, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6004613', 'n_tokens_mistral': 1339, 'n_tokens_neox': 1267, 'n_words': 299}