package com.wenbronk.sparkstreaming.scala.commons

import java.time.Duration

import io.lettuce.core.RedisURI

import io.lettuce.core.cluster.api.StatefulRedisClusterConnection

import io.lettuce.core.cluster.{ClusterClientOptions, ClusterTopologyRefreshOptions, RedisClusterClient}

import scala.collection.immutable

object RedisUtils {

  val ip = "10.110.122.172"

  val ports = Array(, , , , , )

  def getRediceConnect: StatefulRedisClusterConnection[String, String] = {

    val uris: immutable.Seq[RedisURI] = ports.map(port => {

      RedisURI.builder().withHost(ip).withPort(port).build()

    }).toList

    import scala.collection.JavaConverters._

    val redisClusterClient = RedisClusterClient.create(uris.asJava)

    val topologyRefreshOptions = ClusterTopologyRefreshOptions.builder.enablePeriodicRefresh(Duration.ofMinutes()).enableAllAdaptiveRefreshTriggers.build

    redisClusterClient.setOptions(ClusterClientOptions.builder.autoReconnect(true).pingBeforeActivateConnection(true).topologyRefreshOptions(topologyRefreshOptions).build)

    redisClusterClient.connect()

  }

}

package com.wenbronk.sparkstreaming.test

import com.wenbronk.sparkstreaming.scala.commons.RedisUtils

import org.scalatest.FunSuite

class RedisTests extends FunSuite{

  test("redisSet") {

    val utils = RedisUtils

    val connect = utils.getRediceConnect

    connect.async().set("abc", "")

    println("hello")

  }

  test("redisGet") {

    val utils = RedisUtils

    val connect = utils.getRediceConnect

    val value = connect.async().get("abc")

    println(value.get())

  }

  test("flushdb") {

    val utils = RedisUtils

    val connect = utils.getRediceConnect

    val value = connect.async().flushdb()

    println(value.get())

  }

}