Besteleme BodyParser Play 2.5

Question

Bu imza ile bir işlevi göz önüne alındığında?

Basitlik için, bir üstbilginin (belki de "Bazı Başlıklar") gövdeyle tam olarak eşleşen bir değere sahip olduğunu doğrulamak istediğimi varsayalım. Ben bu eylemi var ise: ricam curl -H "Some-Header: hello" -d "hi" http://localhost:9000/post ise ben curl -H "Some-Header: hello" -d "hello" http://localhost:9000/post gibi bir istek o 200 durumundaki tepki vücutta "Merhaba" dönmelidir yaptığınızda

def post(): Action(parser(parse.tolerantText)) { request => 
    Ok(request.body) 
} 

hiçbir vücut ile bir 400 dönmelidir .

İşte denedim. otherParser(request).through(flow), ByteString çıkış için flow beklediğinden, bu bir derleme yapmıyor çünkü bu bir otherParser(request).through(flow). Buradaki fikir, akışın, Either çıkışı üzerinden işleme devam edip etmeyeceği konusunda akümülatörü haberdar edebilmesiydi. Akümülatörün önceki adımın durumunu nasıl bildiğinden emin değilim. Ayrıca filtre kullanmayı denedim. Bu bir derleme yapar ve yazılan cevap gövdesi doğrudur. Ancak, her zaman bir 200 Ok yanıt durumu döndürür.

def parser[A](otherParser: BodyParser[A]): BodyParser[A] = BodyParser { request => 
    val flow: Flow[ByteString, ByteString, akka.NotUsed] = Flow[ByteString].filter { bytes => 
    request.headers.get("Some-Header").contains(bytes.utf8String) 
    } 

    val acc: Accumulator[ByteString, Either[Result, A]] = otherParser(request) 

    acc.through(flow) 
} 

Answer 1

GraphStageWithMaterializedValue kullanarak bir çözüm buldum. Bu konsept Play's maxLength body parser'dan ödünç alınmıştır. Sorgumdaki ilk denemem arasındaki temel fark (derleme yapmayan), akışı mutasyona uğratmak yerine, işleme durumu hakkında bilgi aktarmak için maddi değeri kullanmam gerektiğidir. Bir Flow[ByteString, Either[Result, ByteString], NotUsed] oluşturmuşken, ihtiyacım olan şey Flow[ByteString, ByteString, Future[Boolean]] idi.

def parser[A](otherParser: BodyParser[A]): BodyParser[A] = BodyParser { request => 
    val flow: Flow[ByteString, ByteString, Future[Boolean]] = Flow.fromGraph(new BodyValidator(request.headers.get("Some-Header"))) 

    val parserSink: Sink[ByteString, Future[Either[Result, A]]] = otherParser.apply(request).toSink 

    Accumulator(flow.toMat(parserSink) { (statusFuture: Future[Boolean], resultFuture: Future[Either[Result, A]]) => 
    statusFuture.flatMap { success => 
     if (success) { 
     resultFuture.map { 
      case Left(result) => Left(result) 
      case Right(a) => Right(a) 
     } 
     } else { 
     Future.successful(Left(BadRequest)) 
     } 
    } 
    }) 
} 

anahtar çizgisi bu biridir:

bununla Yani, benim parser fonksiyon bu gibi bakarak biter

val flow: Flow[ByteString, ByteString, Future[Boolean]] = Flow.fromGraph(new BodyValidator(request.headers.get("Some-Header"))) 

dinlenme naziksiniz edebiliyoruz kez yere düşüyor

Bu akışı yarat. Ne yazık ki, BodyValidator oldukça verbose ve biraz kazan-platey hissediyor. Her durumda, okunması oldukça kolaydır. GraphStageWithMaterializedValue, bu grafiğin giriş türünü ve çıkış türünü belirtmek için def shape: S ( burada) uygulamanızı bekler. Ayrıca, grafiğin gerçekte ne yapması gerektiğini tanımlamak için def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, M) (M bir Future[Boolean] buradadır) imlenmesini bekler. İşte BodyValidator tam kodu (aşağıda daha ayrıntılı açıklayacağımız) var:

class BodyValidator(expected: Option[String]) extends GraphStageWithMaterializedValue[FlowShape[ByteString, ByteString], Future[Boolean]] { 
    val in = Inlet[ByteString]("BodyValidator.in") 
    val out = Outlet[ByteString]("BodyValidator.out") 

    override def shape: FlowShape[ByteString, ByteString] = FlowShape.of(in, out) 

    override def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Boolean]) = { 
    val status = Promise[Boolean]() 
    val bodyBuffer = new ByteStringBuilder() 

    val logic = new GraphStageLogic(shape) { 
     setHandler(out, new OutHandler { 
     override def onPull(): Unit = pull(in) 
     }) 

     setHandler(in, new InHandler { 
     def onPush(): Unit = { 
      val chunk = grab(in) 
      bodyBuffer.append(chunk) 
      push(out, chunk) 
     } 

     override def onUpstreamFinish(): Unit = { 
      val fullBody = bodyBuffer.result() 
      status.success(expected.map(ByteString(_)).contains(fullBody)) 
      completeStage() 
     } 

     override def onUpstreamFailure(e: Throwable): Unit = { 
      status.failure(e) 
      failStage(e) 
     } 
     }) 
    } 

    (logic, status.future) 
    } 
} 

Öncelikle grafik için girişler ve çıkışlar kurmak için bir Inlet ve Outlet oluşturmak istediğiniz

val in = Inlet[ByteString]("BodyValidator.in") 
val out = Outlet[ByteString]("BodyValidator.out") 

Sonra bunları shape tanımlamak için kullanırsınız.createLogicAndMaterializedValue İçinde

def shape: FlowShape[ByteString, ByteString] = FlowShape.of(in, out) 

Eğer materialze niyetinde değerini başlatmak gerekir. Burada, akıştan tam verilere sahip olduğumda çözülebilecek bir söz kullandım. Yinelemeler arasındaki verileri izlemek için ByteStringBuilder da oluşturuyorum.

val status = Promise[Boolean]() 
val bodyBuffer = new ByteStringBuilder() 

Sonra bir GraphStageLogic aslında bu grafik işleme her noktada ne yaptığını kurmak oluşturun. İki işleyici ayarlanıyor. Biri, yukarı akış kaynağından gelen verilerle uğraşmak için bir InHandler'dur. Diğeri, aşağı yönde gönderilecek veri ile ilgilenmek için OutHandler'dur. OutHandler'da gerçekten ilginç bir şey yok, dolayısıyla bir IllegalStateException'dan kaçınmak için gerekli kazan plakası olduğunu söylemek için burada görmezden geleceğim. Üç yöntem InHandler: onPush, onUpstreamFinish ve onUpstreamFailure'da geçersiz kılınmıştır. Yeni veriler hazırda olduğunda onPush çağrılır. Bu yöntemde, bir sonraki veri yığınını alıyorum, bodyBuffer'a yazıp veriyi aşağıya doğru aktaracağım.

def onPush(): Unit = { 
    val chunk = grab(in) 
    bodyBuffer.append(chunk) 
    push(out, chunk) 
} 

onUpstreamFinish çağrıldığında üst yüzeyler (sürpriz). Bu, bedeni başlık ile karşılaştırmanın iş mantığının gerçekleştiği yerdir. bir şeyler ters gittiğinde de başarısız olarak, ben hayata geleceği işaretlemek böylece

override def onUpstreamFinish(): Unit = { 
    val fullBody = bodyBuffer.result() 
    status.success(expected.map(ByteString(_)).contains(fullBody)) 
    completeStage() 
} 

onUpstreamFailure

uygulanmaktadır.

override def onUpstreamFailure(e: Throwable): Unit = { 
    status.failure(e) 
    failStage(e) 
} 

Sonra ben sadece bir demet olarak oluşturduğum GraphStageLogic ve status.future dönün.