override def handle()

in quickstart-flink/quickstart-connector/src/main/scala/org/apache/streampark/flink/quickstart/connector/InfluxDBSinkApp.scala [34:49]

• 11 lines of code
• 1 McCabe index (conditional complexity)

  override def handle(): Unit = {
    val source = context.addSource(new WeatherSource())

    //weather,altitude=1000,area=北 temperature=11,humidity=-4

    implicit val entity: InfluxEntity[Weather] = new InfluxEntity[Weather](
      "mydb",
      "test",
      "autogen",
      (x: Weather) => Map("altitude" -> x.altitude.toString, "area" -> x.area),
      (x: Weather) => Map("temperature" -> x.temperature.asInstanceOf[Object], "humidity" -> x.humidity.asInstanceOf[Object])
    )

    InfluxSink().sink(source, "mydb")

  }