Release v0.14.0

Author: José Valim

.travis.yml

@@ -1,14 +1,12 @@
 language: elixir
 elixir:
-  - 1.3.4
-  - 1.4.5
   - 1.5.1
+  - 1.6.5
 sudo: false # to use faster container based build environment
 notifications:
   recipients:
     - [email protected]
 otp_release:
-  - 18.0
   - 19.3
   - 20.0
 script:

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## v0.14.0 (2018-06-10)
+
+This version requires Elixir v1.5+.
+
+### Bug fixes
+
+  * Ensure a `:producer_consumer` stops asking the producer if downstream demand is 0
+  * Enforce the :hash option for non-int partitions in PartitionDispatcher
+
 ## v0.13.1 (2018-01-28)
 
 Note: this is the last version to support Elixir v1.3 and v1.4.

README.md

@@ -24,11 +24,11 @@ Examples for using GenStage and ConsumerSupervisor can be found [examples](examp
 
 ## Installation
 
-GenStage requires Elixir v1.3. Just add `:gen_stage` to your list of dependencies in mix.exs:
+GenStage requires Elixir v1.5. Just add `:gen_stage` to your list of dependencies in mix.exs:
 
 ```elixir
 def deps do
-  [{:gen_stage, "~> 0.13"}]
+  [{:gen_stage, "~> 0.14"}]
 end
 ```
 

lib/consumer_supervisor.ex

@@ -118,21 +118,19 @@ defmodule ConsumerSupervisor do
       @behaviour ConsumerSupervisor
       import Supervisor.Spec
 
-      if Code.ensure_loaded?(Supervisor) and function_exported?(Supervisor, :init, 2) do
-        @doc false
-        def child_spec(arg) do
-          default = %{
-            id: __MODULE__,
-            start: {__MODULE__, :start_link, [arg]},
-            type: :supervisor
-          }
-
-          Supervisor.child_spec(default, unquote(Macro.escape(opts)))
-        end
+      @doc false
+      def child_spec(arg) do
+        default = %{
+          id: __MODULE__,
+          start: {__MODULE__, :start_link, [arg]},
+          type: :supervisor
+        }
 
-        defoverridable child_spec: 1
+        Supervisor.child_spec(default, unquote(Macro.escape(opts)))
       end
 
+      defoverridable child_spec: 1
+
       @doc false
       def init(arg)
     end
@@ -191,8 +189,8 @@ defmodule ConsumerSupervisor do
   name. The supported values are described under the "Name Registration"
   section in the `GenServer` module docs.
   """
-  @spec start_link(module, any) :: Supervisor.on_start
-  @spec start_link(module, any, [option]) :: Supervisor.on_start
+  @spec start_link(module, any) :: Supervisor.on_start()
+  @spec start_link(module, any, [option]) :: Supervisor.on_start()
   def start_link(mod, args, opts \\ []) do
     GenStage.start_link(__MODULE__, {mod, args, opts[:name]}, opts)
   end
@@ -314,16 +312,9 @@ defmodule ConsumerSupervisor do
     {:ok, [template], opts}
   end
 
-  if Code.ensure_loaded?(Supervisor) and function_exported?(Supervisor, :init, 2) do
-    def init([template], opts) when is_tuple(template) or is_map(template) or is_atom(template) do
-      {:ok, {_, [template]}} = Supervisor.init([template], opts)
-      {:ok, [template], opts}
-    end
-  end
-
-  def init(not_a_list, opts) when not is_list(not_a_list) do
-    IO.warn("ConsumerSupervisor.init/1 expects a list with a single element")
-    init([not_a_list], opts)
+  def init([template], opts) when is_tuple(template) or is_map(template) or is_atom(template) do
+    {:ok, {_, [template]}} = Supervisor.init([template], opts)
+    {:ok, [template], opts}
   end
 
   @compile {:inline, call: 2}

lib/gen_stage.ex

@@ -1118,19 +1118,17 @@ defmodule GenStage do
     quote location: :keep, bind_quoted: [opts: opts] do
       @behaviour GenStage
 
-      if Code.ensure_loaded?(Supervisor) and function_exported?(Supervisor, :init, 2) do
-        @doc false
-        def child_spec(arg) do
-          default = %{
-            id: __MODULE__,
-            start: {__MODULE__, :start_link, [arg]}
-          }
-
-          Supervisor.child_spec(default, unquote(Macro.escape(opts)))
-        end
-
-        defoverridable child_spec: 1
+      @doc false
+      def child_spec(arg) do
+        default = %{
+          id: __MODULE__,
+          start: {__MODULE__, :start_link, [arg]}
+        }
+
+        Supervisor.child_spec(default, unquote(Macro.escape(opts)))
       end
+
+      defoverridable child_spec: 1
     end
   end
 

lib/gen_stage/buffer.ex

@@ -7,7 +7,7 @@ defmodule GenStage.Buffer do
   # is stored in a wheel for performance and to avoid discards.
   @moduledoc false
 
-  @opaque t() :: {:queue.queue, non_neg_integer(), wheel()}
+  @opaque t() :: {:queue.queue(), non_neg_integer(), wheel()}
   @typep wheel() :: {non_neg_integer(), pos_integer(), map()} | pos_integer() | reference()
 
   @doc """
@@ -175,20 +175,17 @@ defmodule GenStage.Buffer do
   defp pop_and_increment_wheel({pos, max, wheel}) do
     new_pos = rem(pos + 1, max)
 
-    # TODO: Use :maps.take/2
-    case wheel do
-      %{^pos => perms} when map_size(wheel) == 1 ->
-        {:ok, perms, max}
+    case :maps.take(pos, wheel) do
+      {perms, wheel} ->
+        maybe_triplet = if wheel == %{}, do: max, else: {new_pos, max, wheel}
+        {:ok, perms, maybe_triplet}
 
-      %{^pos => perms} ->
-        {:ok, perms, {new_pos, max, Map.delete(wheel, pos)}}
-
-      %{} ->
+      :error ->
         {:error, {new_pos, max, wheel}}
     end
   end
 
   defp pop_and_increment_wheel(max) do
     {:error, max}
   end
-end
+end

lib/gen_stage/stream.ex

@@ -14,7 +14,8 @@ defmodule GenStage.Stream do
 
   defp stream_validate_opts({to, opts}) when is_list(opts) do
     with {:ok, max, _} <- Utils.validate_integer(opts, :max_demand, 1000, 1, :infinity, false),
-         {:ok, min, _} <- Utils.validate_integer(opts, :min_demand, div(max, 2), 0, max - 1, false),
+         {:ok, min, _} <-
+           Utils.validate_integer(opts, :min_demand, div(max, 2), 0, max - 1, false),
          {:ok, cancel, _} <-
            Utils.validate_in(opts, :cancel, :permanent, [:temporary, :transient, :permanent]) do
       {to, cancel, min, max, opts}

lib/gen_stage/utils.ex

@@ -135,4 +135,4 @@ defmodule GenStage.Utils do
   defp split_events([event | events], limit, counter, acc) do
     split_events(events, limit, counter + 1, [event | acc])
   end
-end
+end

mix.exs

@@ -1,13 +1,13 @@
 defmodule GenStage.Mixfile do
   use Mix.Project
 
-  @version "0.13.1"
+  @version "0.14.0"
 
   def project do
     [
       app: :gen_stage,
       version: @version,
-      elixir: "~> 1.3",
+      elixir: "~> 1.5",
       package: package(),
       description: "Producer and consumer pipelines with back-pressure for Elixir",
       start_permanent: Mix.env() == :prod,

mix.lock

@@ -1,6 +1,6 @@
 %{
-  "earmark": {:hex, :earmark, "1.2.4", "99b637c62a4d65a20a9fb674b8cffb8baa771c04605a80c911c4418c69b75439", [:mix], [], "hexpm"},
-  "ex_doc": {:hex, :ex_doc, "0.18.1", "37c69d2ef62f24928c1f4fdc7c724ea04aecfdf500c4329185f8e3649c915baf", [:mix], [{:earmark, "~> 1.1", [hex: :earmark, repo: "hexpm", optional: false]}], "hexpm"},
+  "earmark": {:hex, :earmark, "1.2.5", "4d21980d5d2862a2e13ec3c49ad9ad783ffc7ca5769cf6ff891a4553fbaae761", [:mix], [], "hexpm"},
+  "ex_doc": {:hex, :ex_doc, "0.18.3", "f4b0e4a2ec6f333dccf761838a4b253d75e11f714b85ae271c9ae361367897b7", [:mix], [{:earmark, "~> 1.1", [hex: :earmark, repo: "hexpm", optional: false]}], "hexpm"},
   "inch_ex": {:hex, :inch_ex, "0.5.5", "b63f57e281467bd3456461525fdbc9e158c8edbe603da6e3e4671befde796a3d", [:mix], [{:poison, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm"},
   "poison": {:hex, :poison, "3.0.0", "625ebd64d33ae2e65201c2c14d6c85c27cc8b68f2d0dd37828fde9c6920dd131", [:mix], [], "hexpm"},
 }