[incubator-seatunnel] branch dev updated: [Improvement][connector-spark-socket] Refactored config arguments and default values (#1687)

[ki...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

kirs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-seatunnel.git


The following commit(s) were added to refs/heads/dev by this push:
     new 1c6b7215 [Improvement][connector-spark-socket] Refactored config arguments and default values (#1687)
1c6b7215 is described below

commit 1c6b7215fbb6282f23fab154e1f7d1d9b9f828c4
Author: mans2singh <ma...@users.noreply.github.com>
AuthorDate: Mon Apr 11 22:35:58 2022 -0400

    [Improvement][connector-spark-socket] Refactored config arguments and default values (#1687)
---
 .../org/apache/seatunnel/spark/socket/Config.scala | 44 ++++++++++++++++++++++
 .../spark/socket/source/SocketStream.scala         |  7 ++--
 2 files changed, 48 insertions(+), 3 deletions(-)

diff --git a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-socket/src/main/scala/org/apache/seatunnel/spark/socket/Config.scala b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-socket/src/main/scala/org/apache/seatunnel/spark/socket/Config.scala
new file mode 100644
index 00000000..23b148ab
--- /dev/null
+++ b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-socket/src/main/scala/org/apache/seatunnel/spark/socket/Config.scala
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seatunnel.spark.socket
+
+/**
+ * SocketStream config parameters and default values
+ */
+object Config extends Serializable {
+
+  /**
+   * Socket host config
+   */
+  val HOST = "host"
+
+  /**
+   * Socket port config
+   */
+  val PORT = "port"
+
+  /**
+   * Default host
+   */
+  val DEFAULT_HOST = "localhost"
+
+  /**
+   * Default port
+   */
+  val DEFAULT_PORT = 9999
+
+}
diff --git a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-socket/src/main/scala/org/apache/seatunnel/spark/socket/source/SocketStream.scala b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-socket/src/main/scala/org/apache/seatunnel/spark/socket/source/SocketStream.scala
index daff716a..b9b9e469 100644
--- a/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-socket/src/main/scala/org/apache/seatunnel/spark/socket/source/SocketStream.scala
+++ b/seatunnel-connectors/seatunnel-connectors-spark/seatunnel-connector-spark-socket/src/main/scala/org/apache/seatunnel/spark/socket/source/SocketStream.scala
@@ -19,6 +19,7 @@ package org.apache.seatunnel.spark.socket.source
 import scala.collection.JavaConversions._
 
 import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory
+import org.apache.seatunnel.spark.socket.Config.{HOST, DEFAULT_HOST, PORT, DEFAULT_PORT}
 import org.apache.seatunnel.spark.SparkEnvironment
 import org.apache.seatunnel.spark.stream.SparkStreamingSource
 import org.apache.spark.rdd.RDD
@@ -31,13 +32,13 @@ class SocketStream extends SparkStreamingSource[String] {
   override def prepare(env: SparkEnvironment): Unit = {
     val defaultConfig = ConfigFactory.parseMap(
       Map(
-        "host" -> "localhost",
-        "port" -> 9999))
+        HOST -> DEFAULT_HOST,
+        PORT -> DEFAULT_PORT))
     config = config.withFallback(defaultConfig)
   }
 
   override def getData(env: SparkEnvironment): DStream[String] = {
-    env.getStreamingContext.socketTextStream(config.getString("host"), config.getInt("port"))
+    env.getStreamingContext.socketTextStream(config.getString(HOST), config.getInt(PORT))
   }
 
   override def rdd2dataset(sparkSession: SparkSession, rdd: RDD[String]): Dataset[Row] = {