fix elastic transports (#17)

* fix elastic transports

* add tests

* update readme skip ci

README.md

@@ -43,7 +43,7 @@
 `Flinkrunner 3` is now on maven central, built against Flink 1.11 with Scala 2.12 and JDK 11.
 
 ```sbtshell
-libraryDependencies += "io.epiphanous" %% "flinkrunner" % "3.0.0"
+libraryDependencies += "io.epiphanous" %% "flinkrunner" % "3.0.3"
 ```
 
 ## What is FlinkRunner?

src/main/scala/io/epiphanous/flinkrunner/util/StreamUtils.scala

@@ -498,10 +498,10 @@ object StreamUtils extends LazyLogging {
   ) = {
     val hosts = sinkConfig.transports
       .map(s => {
-        val url = new URL(s"https://${s}")
+        val url = new URL(if (s.startsWith("http")) s else s"http://${s}")
         val hostname = url.getHost
         val port = if (url.getPort < 0) 9200 else url.getPort
-        new HttpHost(hostname, port, "https")
+        new HttpHost(hostname, port, url.getProtocol)
       })
       .asJava
     val esSink = new ElasticsearchSink.Builder[E](hosts, new ElasticsearchSinkFunction[E] {

src/test/scala/io/epiphanous/flinkrunner/model/ElasticsearchSinkConfigSpec.scala

@@ -0,0 +1,27 @@
+package io.epiphanous.flinkrunner.model
+
+import java.net.URL
+import java.util.Properties
+
+import io.epiphanous.flinkrunner.BasePropSpec
+import org.apache.http.HttpHost
+
+import scala.collection.JavaConverters._
+
+class ElasticsearchSinkConfigSpec extends BasePropSpec {
+  def getHosts(transports:List[String]) = transports
+    .map(s => {
+      val url = new URL(if (s.startsWith("http")) s else s"http://${s}")
+      val hostname = url.getHost
+      val port = if (url.getPort < 0) 9200 else url.getPort
+      new HttpHost(hostname, port, url.getProtocol)
+    })
+    .asJava
+
+  property("transports config with protocol") {
+    println(getHosts(List("https://elastic:9200")))
+  }
+  property("transports config without protocol") {
+    println(getHosts(List("elastic:9200")))
+  }
+}