feat(conf): add some conf utils

src/main/scala/com/zdjizhi/flink/voip/conf/FusionConfigs.scala

@@ -0,0 +1,48 @@
+package com.zdjizhi.flink.voip.conf
+
+import org.apache.flink.api.common.time.Time
+import org.apache.flink.configuration.{ConfigOption, ConfigOptions}
+
+import java.lang.{Long => JLong}
+
+/**
+ * An object containing configuration options for the Fusion application.
+ *
+ * @author chaoc
+ * @since 1.0
+ */
+object FusionConfigs {
+
+  val SOURCE_KAFKA_TOPIC: ConfigOption[String] =
+    ConfigOptions.key("source.kafka.topic")
+      .stringType()
+      .noDefaultValue()
+      .withDescription("")
+
+  val SINK_KAFKA_TOPIC: ConfigOption[String] =
+    ConfigOptions.key("sink.kafka.topic")
+      .stringType()
+      .noDefaultValue()
+      .withDescription("")
+
+  val SOURCE_KAFKA_PROPERTIES_PREFIX: String = "source.kafka.props."
+
+  val SINK_KAFKA_PROPERTIES_PREFIX: String = "sink.kafka.props."
+
+  /**
+   * The configuration option for the interval at which SIP (Session Initiation Protocol) state data
+   * should be cleared.
+   */
+  val SIP_STATE_CLEAR_INTERVAL: ConfigOption[JLong] =
+    ConfigOptions.key("")
+      .longType()
+      .defaultValue(Time.minutes(1).toMilliseconds)
+      .withDescription("")
+
+  val RTP_STATE_CLEAR_INTERVAL: ConfigOption[JLong] =
+    ConfigOptions.key("")
+      .longType()
+      .defaultValue(Time.minutes(3).toMilliseconds)
+      .withDescription("")
+
+}

src/main/scala/com/zdjizhi/flink/voip/conf/FusionParameterTool.scala

@@ -0,0 +1,32 @@
+package com.zdjizhi.flink.voip.conf
+
+import org.apache.flink.api.java.utils.ParameterTool
+
+import java.util.Properties
+import scala.collection.JavaConverters._
+
+/**
+ * A wrapper class that extends the Flink `ParameterTool` to provide utility methods for handling
+ * properties with a specific prefix. This class allows retrieving properties that start with the
+ * given `prefix` and converts them into a `java.util.Properties` object.
+ *
+ * @param tool The original Flink `ParameterTool` instance.
+ * @author chaoc
+ * @since 1.0
+ */
+class FusionParameterTool(tool: ParameterTool) {
+
+  /**
+   * Retrieves properties from the underlying `ParameterTool` instance that start with the specified
+   * `prefix`. The properties are then converted into a `java.util.Properties` object and returned.
+   *
+   * @param prefix The prefix to filter properties.
+   * @return A `java.util.Properties` object containing the properties with the specified prefix.
+   */
+  def getProperties(prefix: String): Properties = {
+    val map = tool.toMap.asScala.filterKeys(_.startsWith(prefix))
+      .map { case (key, value) => (key.stripPrefix(prefix), value) }
+    ParameterTool.fromMap(map.asJava).getProperties
+  }
+
+}

src/main/scala/com/zdjizhi/flink/voip/conf/package.scala

@@ -0,0 +1,11 @@
+package com.zdjizhi.flink.voip
+
+import org.apache.flink.api.java.utils.ParameterTool
+
+import scala.language.implicitConversions
+
+package object conf {
+
+  implicit def asFusionParameterTool(tool: ParameterTool): FusionParameterTool = new FusionParameterTool(tool)
+
+}