如何使用Akka HTTP解编集json响应以删除不必要的字段

svgewumm  于 2022-11-05  发布在  其他
关注(0)|答案(1)|浏览(173)

我是AkkaHTTP的新手,我想去掉JSON响应中不必要的字段,只取必要的字段。我使用this端点来获取响应,它包含了一堆字段。目前我只需要“name”和“versions”。我想知道如何将其反序列化为只包含“name”和“versions”的case类。我编写了下面几行代码以获得字符串形式的响应。

import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.stream.scaladsl.{Flow, Sink, Source}
import akka.stream.{ActorMaterializer, OverflowStrategy}

import scala.concurrent.Future
import scala.concurrent.duration.DurationInt
import scala.language.postfixOps
import scala.util.{Failure, Success}

object SoftwareRegistry extends App {

  implicit val system = ActorSystem("NPMRegistry")
  implicit val materializer = ActorMaterializer()

  import system.dispatcher

  case class NPMPackage(name: String)

  // reading the packages
  val filename = "B:\\Scala\\NPMRegistry\\src\\main\\resources\\packages.txt"
  val bufferedSource = scala.io.Source.fromFile(filename)
  val listOfPackages: List[NPMPackage] = (for (line <- bufferedSource.getLines) yield {
    NPMPackage(line.trim)
  }).toList
  bufferedSource.close()

  // source
  val sourceList = Source(listOfPackages)

  // sink
  val sink = Sink.foreach[NPMPackage] { p =>
    // https request
    val responseFuture: Future[HttpResponse] =
      Http().singleRequest(HttpRequest(uri = s"https://registry.npmjs.org/${p.name}"))
    val x = responseFuture
      .flatMap(_.entity.toStrict(2 seconds))
      .map(_.data.utf8String)
    x.onComplete {
      case Success(res) => println(res)
      case Failure(_) => sys.error("Something went wrong")
    }
  }

  // flow to slow things down and streaming sink to time-delayed operations
  val bufferedFlow = Flow[NPMPackage]
    .buffer(10, overflowStrategy = OverflowStrategy.backpressure)
    .throttle(1, 3 seconds)

  sourceList.async
    .via(bufferedFlow).async
    .to(sink)
    .run()
}

它会输出以下输出

eblbsuwk

eblbsuwk1#

为了解析json,你需要使用一些库。在akka-http文档中,它们使用spray-json。用适当的akkaHttpVersion将下面的依赖项添加到你的build.sbt

"com.typesafe.akka" %% "akka-http-spray-json" % akkaHttpVersion

现在你需要序列化器和反序列化器来处理你的数据。我使用的是一个简单的模型,根据需要进行修改。

trait Formatter extends DefaultJsonProtocol {

  implicit object jsonFormat extends JsonFormat[Versions] {
    override def read(json: JsValue): Versions = json match {
      case JsObject(fields) =>
        Versions(fields.keys.toList)
    }

    override def write(obj: Versions): JsValue = JsonParser(obj.toString)
  }

  implicit val formatterPackage: RootJsonFormat[Package] = jsonFormat2(Package)

  case class Package(name: String, versions: Versions)

  case class Versions(versions: List[String])
}

最后sink

//needed import with others
 import spray.json._

 object SoftwareRegistry extends App  with Formatter {

   //existing code
   //---------

   val sink = Sink.foreach[NPMPackage] { p =>
       // https request
       val responseFuture: Future[HttpResponse] =
         Http().singleRequest(HttpRequest(uri = s"https://registry.npmjs.org/${p.name}"))
       val packages = responseFuture
         .flatMap(
           _.entity
             .dataBytes
             .via(JsonFraming.objectScanner(Int.MaxValue))
             .map(_.utf8String)
             .map(_.parseJson.convertTo[Package])
             .toMat(Sink.seq)(Keep.right)
             .run()
         )

       packages.onComplete {
         case Success(res) => println(res)
         case Failure(_) => sys.error("Something went wrong")
       }
   }

   //existing code
   //---------
}

相关问题