Cleaning up and formatting code

Author: dsrees

src/main/kotlin/org/phoenixframework/Defaults.kt

@@ -38,21 +38,22 @@ object Defaults {
 
   /** Default reconnect algorithm for the socket */
   val reconnectSteppedBackOff: (Int) -> Long = { tries ->
-    if (tries > 9) 5_000 else listOf(10L, 50L, 100L, 150L, 200L, 250L, 500L, 1_000L, 2_000L)[tries - 1]
+    if (tries > 9) 5_000 else listOf(
+      10L, 50L, 100L, 150L, 200L, 250L, 500L, 1_000L, 2_000L
+    )[tries - 1]
   }
 
   /** Default rejoin algorithm for individual channels */
   val rejoinSteppedBackOff: (Int) -> Long = { tries ->
     if (tries > 3) 10_000 else listOf(1_000L, 2_000L, 5_000L)[tries - 1]
   }
 
-
   /** The default Gson configuration to use when parsing messages */
   val gson: Gson
     get() = GsonBuilder()
-        .setLenient()
-        .setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES)
-        .create()
+      .setLenient()
+      .setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES)
+      .create()
 
   /**
    * Takes an endpoint and a params closure given by the User and constructs a URL that
@@ -62,7 +63,10 @@ object Defaults {
    *
    * @throws IllegalArgumentException if [endpoint] is not a valid URL endpoint.
    */
-  internal fun buildEndpointUrl(endpoint: String, paramsClosure: PayloadClosure?): URL {
+  internal fun buildEndpointUrl(
+    endpoint: String,
+    paramsClosure: PayloadClosure
+  ): URL {
     var mutableUrl = endpoint
     // Silently replace web socket URLs with HTTP URLs.
     if (endpoint.regionMatches(0, "ws:", 0, 3, ignoreCase = true)) {
@@ -72,8 +76,9 @@ object Defaults {
     }
 
     // If there are query params, append them now
-    var httpUrl = HttpUrl.parse(mutableUrl) ?: throw IllegalArgumentException("invalid url: $endpoint")
-    paramsClosure?.invoke()?.let {
+    var httpUrl =
+      HttpUrl.parse(mutableUrl) ?: throw IllegalArgumentException("invalid url: $endpoint")
+    paramsClosure.invoke()?.let {
       val httpBuilder = httpUrl.newBuilder()
       it.forEach { (key, value) ->
         httpBuilder.addQueryParameter(key, value.toString())
@@ -85,6 +90,4 @@ object Defaults {
     // Store the URL that will be used to establish a connection
     return httpUrl.url()
   }
-
-
 }

src/main/kotlin/org/phoenixframework/Socket.kt

@@ -45,22 +45,34 @@ internal class StateChangeCallbacks {
     private set
 
   /** Safely adds an onOpen callback */
-  fun onOpen(ref: String, callback: () -> Unit) {
+  fun onOpen(
+    ref: String,
+    callback: () -> Unit
+  ) {
     this.open = this.open + Pair(ref, callback)
   }
 
   /** Safely adds an onClose callback */
-  fun onClose(ref: String, callback: () -> Unit) {
+  fun onClose(
+    ref: String,
+    callback: () -> Unit
+  ) {
     this.close = this.close + Pair(ref, callback)
   }
 
   /** Safely adds an onError callback */
-  fun onError(ref: String, callback: (Throwable, Response?) -> Unit) {
+  fun onError(
+    ref: String,
+    callback: (Throwable, Response?) -> Unit
+  ) {
     this.error = this.error + Pair(ref, callback)
   }
 
   /** Safely adds an onMessage callback */
-  fun onMessage(ref: String, callback: (Message) -> Unit) {
+  fun onMessage(
+    ref: String,
+    callback: (Message) -> Unit
+  ) {
     this.message = this.message + Pair(ref, callback)
   }
 
@@ -96,7 +108,6 @@ typealias PayloadClosure = () -> Payload?
  * Connects to a Phoenix Server
  */
 
-
 /**
  * A [Socket] which connects to a Phoenix Server. Takes a closure to allow for changing parameters
  * to be sent to the server when connecting.
@@ -112,7 +123,7 @@ typealias PayloadClosure = () -> Payload?
  */
 class Socket(
   url: String,
-  paramsClosure: PayloadClosure?,
+  val paramsClosure: PayloadClosure,
   private val gson: Gson = Defaults.gson,
   private val client: OkHttpClient = OkHttpClient.Builder().build()
 ) {
@@ -132,12 +143,6 @@ class Socket(
   var endpointUrl: URL
     private set
 
-  /**
-   * A closure that returns the optional params to pass when connecting. Must
-   * be set when initializing the Socket. These will be appended to the URL.
-   */
-  val paramsClosure: PayloadClosure? = paramsClosure
-
   /** Timeout to use when opening a connection */
   var timeout: Long = Defaults.TIMEOUT
 
@@ -229,7 +234,7 @@ class Socket(
     params: Payload? = null,
     gson: Gson = Defaults.gson,
     client: OkHttpClient = OkHttpClient.Builder().build()
-  ): this(url, params?.let { { it }  }, gson, client)
+  ) : this(url, { params }, gson, client)
 
   init {
     var mutableUrl = url
@@ -245,7 +250,6 @@ class Socket(
       mutableUrl += "websocket"
     }
 
-
     // Store the endpoint before changing the protocol
     this.endpoint = mutableUrl
 
@@ -255,12 +259,12 @@ class Socket(
 
     // Create reconnect timer
     this.reconnectTimer = TimeoutTimer(
-        dispatchQueue = dispatchQueue,
-        timerCalculation = reconnectAfterMs,
-        callback = {
-          this.logItems("Socket attempting to reconnect")
-          this.teardown { this.connect() }
-        })
+      dispatchQueue = dispatchQueue,
+      timerCalculation = reconnectAfterMs,
+      callback = {
+        this.logItems("Socket attempting to reconnect")
+        this.teardown { this.connect() }
+      })
   }
 
   //------------------------------------------------------------------------------
@@ -289,8 +293,7 @@ class Socket(
     this.closeWasClean = false
 
     // Build the new endpointUrl with the params closure. The payload returned
-    // from the closure could have changed after the socket attempts to reconnect,
-    // i.e. and authToken.
+    // from the closure could be different such as a changing authToken.
     this.endpointUrl = Defaults.buildEndpointUrl(this.endpoint, this.paramsClosure)
 
     // Now create the connection transport and attempt to connect
@@ -313,7 +316,6 @@ class Socket(
     // Reset any reconnects and teardown the socket connection
     this.reconnectTimer.reset()
     this.teardown(code, reason, callback)
-
   }
 
   fun onOpen(callback: (() -> Unit)): String {
@@ -336,7 +338,10 @@ class Socket(
     this.stateChangeCallbacks.release()
   }
 
-  fun channel(topic: String, params: Payload = mapOf()): Channel {
+  fun channel(
+    topic: String,
+    params: Payload = mapOf()
+  ): Channel {
     val channel = Channel(topic, params, this)
     this.channels = this.channels + channel
 
@@ -350,7 +355,7 @@ class Socket(
     // removed instead of calling .remove() on the list, thus returning a new list
     // that does not contain the channel that was removed.
     this.channels = channels
-        .filter { it.joinRef != channel.joinRef }
+      .filter { it.joinRef != channel.joinRef }
   }
 
   /**
@@ -481,7 +486,7 @@ class Socket(
     val period = heartbeatIntervalMs
 
     heartbeatTask =
-        dispatchQueue.queueAtFixedRate(delay, period, TimeUnit.MILLISECONDS) { sendHeartbeat() }
+      dispatchQueue.queueAtFixedRate(delay, period, TimeUnit.MILLISECONDS) { sendHeartbeat() }
   }
 
   internal fun sendHeartbeat() {
@@ -503,10 +508,11 @@ class Socket(
     // The last heartbeat was acknowledged by the server. Send another one
     this.pendingHeartbeatRef = this.makeRef()
     this.push(
-        topic = "phoenix",
-        event = Channel.Event.HEARTBEAT.value,
-        payload = mapOf(),
-        ref = pendingHeartbeatRef)
+      topic = "phoenix",
+      event = Channel.Event.HEARTBEAT.value,
+      payload = mapOf(),
+      ref = pendingHeartbeatRef
+    )
   }
 
   private fun abnormalClose(reason: String) {
@@ -570,14 +576,17 @@ class Socket(
 
     // Dispatch the message to all channels that belong to the topic
     this.channels
-        .filter { it.isMember(message) }
-        .forEach { it.trigger(message) }
+      .filter { it.isMember(message) }
+      .forEach { it.trigger(message) }
 
     // Inform all onMessage callbacks of the message
     this.stateChangeCallbacks.message.forEach { it.second.invoke(message) }
   }
 
-  internal fun onConnectionError(t: Throwable, response: Response?) {
+  internal fun onConnectionError(
+    t: Throwable,
+    response: Response?
+  ) {
     this.logItems("Transport: error $t")
 
     // Send an error to all channels
@@ -586,5 +595,4 @@ class Socket(
     // Inform any state callbacks of the error
     this.stateChangeCallbacks.error.forEach { it.second.invoke(t, response) }
   }
-
 }

src/test/kotlin/org/phoenixframework/SocketTest.kt

@@ -48,7 +48,7 @@ class SocketTest {
     internal fun `sets defaults`() {
       val socket = Socket("wss://localhost:4000/socket")
 
-      assertThat(socket.paramsClosure).isNull()
+      assertThat(socket.paramsClosure.invoke()).isNull()
       assertThat(socket.channels).isEmpty()
       assertThat(socket.sendBuffer).isEmpty()
       assertThat(socket.ref).isEqualTo(0)