Merge branch 'fix-warnings-for-1.4'

Author: MaxPower15

lib/nsq/connection.ex

@@ -143,10 +143,11 @@ defmodule NSQ.Connection do
 
 
   @spec handle_cast(:reconnect, state) :: {:noreply, state}
+  def handle_cast(:reconnect, %{connect_attempts: connect_attempts} = conn_state) when connect_attempts > 0 do
+    {_, conn_state} = Initializer.connect(conn_state)
+    {:noreply, conn_state}
+  end
   def handle_cast(:reconnect, conn_state) do
-    if conn_state.connect_attempts > 0 do
-      {_, conn_state} = Initializer.connect(conn_state)
-    end
     {:noreply, conn_state}
   end
 

lib/nsq/connection/buffer.ex

@@ -208,9 +208,12 @@ defmodule NSQ.Connection.Buffer do
           end
       end
 
-    if is_list(decompressed) do
-      decompressed = decompressed |> List.flatten |> Enum.join("")
-    end
+    decompressed =
+      if is_list(decompressed) do
+        decompressed |> List.flatten |> Enum.join("")
+      else
+        decompressed
+      end
 
     combined_buffer = state.buffered_data <> decompressed
 

lib/nsq/connection/initializer.ex

@@ -134,26 +134,27 @@ defmodule NSQ.Connection.Initializer do
     end
 
     # respect negotiated msg_timeout
-    if parsed["msg_timeout"] do
-      conn_state = %{conn_state | msg_timeout: parsed["msg_timeout"]}
-    else
-      conn_state = %{conn_state | msg_timeout: conn_state.config.msg_timeout}
-    end
+    timeout = parsed["msg_timeout"] || conn_state.config.msg_timeout
+    conn_state = %{conn_state | msg_timeout: timeout}
 
     # wrap our socket with SSL if TLS is enabled
-    if parsed["tls_v1"] == true do
-      Logger.debug "Upgrading to TLS..."
-      socket = Socket.SSL.connect! conn_state.socket, [
-        cacertfile: conn_state.config.tls_cert,
-        keyfile: conn_state.config.tls_key,
-        versions: ssl_versions(conn_state.config.tls_min_version),
-        verify: ssl_verify_atom(conn_state.config),
-      ]
-      conn_state = %{conn_state | socket: socket}
-      conn_state.reader |> Buffer.setup_socket(socket, conn_state.config.read_timeout)
-      conn_state.writer |> Buffer.setup_socket(socket, conn_state.config.read_timeout)
-      conn_state |> wait_for_ok!
-    end
+    conn_state =
+      if parsed["tls_v1"] == true do
+        Logger.debug "Upgrading to TLS..."
+        socket = Socket.SSL.connect! conn_state.socket, [
+          cacertfile: conn_state.config.tls_cert,
+          keyfile: conn_state.config.tls_key,
+          versions: ssl_versions(conn_state.config.tls_min_version),
+          verify: ssl_verify_atom(conn_state.config),
+        ]
+        conn_state = %{conn_state | socket: socket}
+        conn_state.reader |> Buffer.setup_socket(socket, conn_state.config.read_timeout)
+        conn_state.writer |> Buffer.setup_socket(socket, conn_state.config.read_timeout)
+        conn_state |> wait_for_ok!
+        conn_state
+      else
+        conn_state
+      end
 
     # If compression is enabled, we expect to receive a compressed "OK"
     # immediately.

lib/nsq/connection/message_handling.ex

@@ -29,28 +29,29 @@ defmodule NSQ.Connection.MessageHandling do
     end
   end
 
+  def handle_nsq_message({:response, "_heartbeat_"}, state) do
+    respond_to_heartbeat(state)
+    {:ok, state}
+  end
 
-  def handle_nsq_message(msg, state) do
-    case msg do
-      {:response, "_heartbeat_"} ->
-        respond_to_heartbeat(state)
-
-      {:response, data} ->
-        {:ok, state} = state |> Command.send_response_to_caller(data)
-
-      {:error, data} ->
-        state |> log_error(nil, data)
-
-      {:error, reason, data} ->
-        state |> log_error(reason, data)
+  def handle_nsq_message({:response, data}, state) do
+    state |> Command.send_response_to_caller(data)
+  end
 
-      {:message, data} ->
-        {:ok, state} = state |> kick_off_message_processing(data)
-    end
+  def handle_nsq_message({:error, data}, state) do
+    state |> log_error(nil, data)
+    {:ok, state}
+  end
 
+  def handle_nsq_message({:error, reason, data}, state) do
+    state |> log_error(reason, data)
     {:ok, state}
   end
 
+  def handle_nsq_message({:message, data}, state) do
+    state |> kick_off_message_processing(data)
+  end
+
 
   @spec update_conn_stats_on_message_done(C.state, any) :: any
   def update_conn_stats_on_message_done(state, ret_val) do

lib/nsq/consumer/backoff.ex

@@ -45,32 +45,31 @@ defmodule NSQ.Consumer.Backoff do
   start_stop_continue.)
   """
   @spec resume(pid, C.state) :: {:ok, C.state}
+  def resume(_cons, %{backoff_duration: 0, backoff_counter: 0} = cons_state),
+    # looks like we successfully left backoff mode already
+    do: {:ok, cons_state}
+
+  def resume(_cons, %{stop_flag: true} = cons_state),
+    do: {:ok, %{cons_state | backoff_duration: 0}}
+
   def resume(cons, cons_state) do
-    if cons_state.backoff_duration == 0 || cons_state.backoff_counter == 0 do
-      # looks like we successfully left backoff mode already
-      {:ok, cons_state}
-    else
-      if cons_state.stop_flag do
-        {:ok, %{cons_state | backoff_duration: 0}}
+    {:ok, cons_state} =
+      if Connections.count(cons_state) == 0 do
+        # This could happen if nsqlookupd suddenly stops discovering
+        # connections. Maybe a network partition?
+        Logger.warn("no connection available to resume")
+        Logger.warn("backing off for 1 second")
+        resume_later(cons, 1000, cons_state)
       else
-        if Connections.count(cons_state) == 0 do
-          # This could happen if nsqlookupd suddenly stops discovering
-          # connections. Maybe a network partition?
-          Logger.warn("no connection available to resume")
-          Logger.warn("backing off for 1 second")
-          {:ok, cons_state} = resume_later(cons, 1000, cons_state)
-        else
-          # pick a random connection to test the waters
-          conn = random_connection_for_backoff(cons_state)
-          Logger.warn("(#{inspect conn}) backoff timeout expired, sending RDY 1")
+        # pick a random connection to test the waters
+        conn = random_connection_for_backoff(cons_state)
+        Logger.warn("(#{inspect conn}) backoff timeout expired, sending RDY 1")
 
-          # while in backoff only ever let 1 message at a time through
-          {:ok, cons_state} = RDY.update(cons, conn, 1, cons_state)
-        end
-
-        {:ok, %{cons_state | backoff_duration: 0}}
+        # while in backoff only ever let 1 message at a time through
+        RDY.update(cons, conn, 1, cons_state)
       end
-    end
+
+    {:ok, %{cons_state | backoff_duration: 0}}
   end
 
   def resume!(cons, cons_state) do

lib/nsq/consumer/connections.ex

@@ -134,10 +134,14 @@ defmodule NSQ.Consumer.Connections do
       # `RDY.redistribute` loop will take care of getting this connection some
       # messages later.
       remaining_rdy = cons_state.max_in_flight - total_rdy_count(cons_state)
-      if remaining_rdy > 0 do
-        conn = conn_from_nsqd(cons, nsqd, cons_state)
-        {:ok, cons_state} = RDY.transmit(conn, 1, cons_state)
-      end
+      cons_state =
+        if remaining_rdy > 0 do
+          conn = conn_from_nsqd(cons, nsqd, cons_state)
+          {:ok, cons_state} = RDY.transmit(conn, 1, cons_state)
+          cons_state
+        else
+          cons_state
+        end
 
       {:ok, cons_state}
     catch

lib/nsq/consumer/rdy.ex

@@ -83,20 +83,25 @@ defmodule NSQ.Consumer.RDY do
     # Cap the given RDY based on how much we can actually assign. Unless it's
     # 0, in which case we'll be retrying.
     max_possible_rdy = calc_max_possible(cons_state, conn_info)
-    if max_possible_rdy > 0 do
-      new_rdy = [new_rdy, max_possible_rdy] |> Enum.min |> round
-    end
-
-    if max_possible_rdy <= 0 && new_rdy > 0 do
-      if conn_info.rdy_count == 0 do
-        # Schedule RDY.update(consumer, conn, new_rdy) for this connection
-        # again in 5 seconds. This is to prevent eternal starvation.
-        {:ok, cons_state} = retry(cons, conn, new_rdy, cons_state)
+    new_rdy = if max_possible_rdy > 0 do
+                [new_rdy, max_possible_rdy] |> Enum.min |> round
+              else
+                new_rdy
+              end
+
+    {:ok, cons_state} =
+      if max_possible_rdy <= 0 && new_rdy > 0 do
+        if conn_info.rdy_count == 0 do
+          # Schedule RDY.update(consumer, conn, new_rdy) for this connection
+          # again in 5 seconds. This is to prevent eternal starvation.
+          retry(cons, conn, new_rdy, cons_state)
+        else
+          {:ok, cons_state}
+        end
+      else
+        transmit(conn, new_rdy, cons_state)
       end
-      {:ok, cons_state}
-    else
-      {:ok, _cons_state} = transmit(conn, new_rdy, cons_state)
-    end
+    {:ok, cons_state}
   end
 
   def update!(cons, conn, new_rdy, cons_state) do

lib/nsq/lookupd.ex

@@ -58,9 +58,7 @@ defmodule NSQ.Lookupd do
 
   @spec normalize_200_response([any], binary) :: response
   defp normalize_200_response(headers, body) do
-    if body == nil || body == "" do
-      body = "{}"
-    end
+    body = if body == nil || body == "", do: "{}", else: body
 
     if headers[:"X-Nsq-Content-Type"] == "nsq; version=1.0" do
       Poison.decode!(body)

lib/nsq/message.ex

@@ -110,19 +110,25 @@ defmodule NSQ.Message do
   mode, where it stops receiving messages for a fixed duration.
   """
   def req(message, delay \\ -1, backoff \\ false) do
-    if delay == -1 do
-      delay = calculate_delay(
-        message.attempts, message.config.max_requeue_delay
-      )
-      Logger.debug("(#{inspect message.connection}) requeue msg ID #{message.id}, delay #{delay} (auto-calculated with attempts #{message.attempts}), backoff #{backoff}")
-    else
-      Logger.debug("(#{inspect message.connection}) requeue msg ID #{message.id}, delay #{delay}, backoff #{backoff}")
-    end
+    delay =
+      if delay == -1 do
+        delay = calculate_delay(
+          message.attempts, message.config.max_requeue_delay
+        )
+        Logger.debug("(#{inspect message.connection}) requeue msg ID #{message.id}, delay #{delay} (auto-calculated with attempts #{message.attempts}), backoff #{backoff}")
+        delay
+      else
+        Logger.debug("(#{inspect message.connection}) requeue msg ID #{message.id}, delay #{delay}, backoff #{backoff}")
+        delay
+      end
 
-    if delay > message.config.max_requeue_delay do
-      Logger.warn "Invalid requeue delay #{delay}. Must be between 0 and #{message.config.max_requeue_delay}. Sending with max delay #{message.config.max_requeue_delay} instead."
-      delay = message.config.max_requeue_delay
-    end
+    delay =
+      if delay > message.config.max_requeue_delay do
+        Logger.warn "Invalid requeue delay #{delay}. Must be between 0 and #{message.config.max_requeue_delay}. Sending with max delay #{message.config.max_requeue_delay} instead."
+        message.config.max_requeue_delay
+      else
+        delay
+      end
 
     if backoff do
       GenServer.call(message.consumer, {:start_stop_continue_backoff, :backoff})