Add Stream.uniq/2

Author: José Valim

lib/elixir/lib/enum.ex

@@ -1520,7 +1520,7 @@ defmodule Enum do
   end
 
   @doc """
-  Iterates the enumerable, removing all duplicated items.
+  Enumerates the collection, removing all duplicated items.
 
   ## Examples
 
@@ -1540,16 +1540,8 @@ defmodule Enum do
   end
 
   def uniq(collection, fun) do
-    { list, _ } =
-      reduce(collection, { [], [] }, fn(entry, { acc, fun_acc }) ->
-        fun_entry = fun.(entry)
-        if :lists.member(fun_entry, fun_acc) do
-          { acc, fun_acc }
-        else
-          { [entry|acc], [fun_entry|fun_acc] }
-        end
-      end)
-
+    { _, { list, _ } } =
+      Enumerable.reduce(collection, { :cont, { [], [] } }, R.uniq(fun))
     :lists.reverse(list)
   end
 

lib/elixir/lib/stream.ex

@@ -544,6 +544,29 @@ defmodule Stream do
     lazy enum, fn(f1) -> R.take_while(fun, f1) end
   end
 
+  @doc """
+  Creates a stream that only emits elements if they are uniq.
+
+  Keep in mind that, in order to know if an element is unique
+  or not, this function needs to store all uniq values emitted
+  by the stream. Therefore, if the stream is infinite, the number
+  of items stored will grow infinitely, never being garbage collected.
+
+  ## Examples
+
+      iex> Stream.uniq([1, 2, 3, 2, 1]) |> Enum.to_list
+      [1, 2, 3]
+
+      iex> Stream.uniq([{1, :x}, {2, :y}, {1, :z}], fn {x, _} -> x end) |> Enum.to_list
+      [{1,:x}, {2,:y}]
+
+  """
+  @spec uniq(Enumerable.t) :: Enumerable.t
+  @spec uniq(Enumerable.t, (element -> term)) :: Enumerable.t
+  def uniq(enum, fun // fn x -> x end) do
+    lazy enum, [], fn f1 -> R.uniq(fun, f1) end
+  end
+
   @doc """
   Creates a stream where each item in the enumerable will
   be accompanied by its index.

lib/elixir/lib/stream/reducers.ex

@@ -25,18 +25,18 @@ defmodule Stream.Reducers do
     end
   end
 
-  defmacro chunks_by(fun, f // nil) do
+  defmacro chunks_by(callback, f // nil) do
     quote do
       fn
         entry, acc(h, { buffer, value }, t) ->
-          new_value = unquote(fun).(entry)
+          new_value = unquote(callback).(entry)
           if new_value == value do
             { :cont, acc(h, { [entry|buffer], value }, t) }
           else
             cont_with_acc(unquote(f), :lists.reverse(buffer), h, { [entry], new_value }, t)
           end
         entry, acc(h, nil, t) ->
-          { :cont, acc(h, { [entry], unquote(fun).(entry) }, t) }
+          { :cont, acc(h, { [entry], unquote(callback).(entry) }, t) }
       end
     end
   end
@@ -144,6 +144,19 @@ defmodule Stream.Reducers do
     end
   end
 
+  defmacro uniq(callback, f // nil) do
+    quote do
+      fn(entry, acc(h, prev, t) = acc) ->
+        value = unquote(callback).(entry)
+        if :lists.member(value, prev) do
+          { :cont, acc }
+        else
+          cont_with_acc(unquote(f), entry, h, [value|prev], t)
+        end
+      end
+    end
+  end
+
   defmacro with_index(f // nil) do
     quote do
       fn(entry, acc(h, counter, t)) ->

lib/elixir/test/elixir/stream_test.exs

@@ -366,6 +366,14 @@ defmodule StreamTest do
     assert Enum.zip(list, list) == Enum.zip(stream, stream)
   end
 
+  test "uniq" do
+    assert Stream.uniq([1, 2, 3, 2, 1]) |> Enum.to_list ==
+           [1, 2, 3]
+
+    assert Stream.uniq([{1, :x}, {2, :y}, {1, :z}], fn {x, _} -> x end) |> Enum.to_list ==
+           [{1,:x}, {2,:y}]
+  end
+
   test "zip/2" do
     concat = Stream.concat(1..3, 4..6)
     cycle  = Stream.cycle([:a, :b, :c])