     test "exits on noconnection :DOWN in message queue" do
       task = create_dummy_task(:noconnection)
       send(self(), {:DOWN, task.ref, :process, task.pid, :noconnection})
+
       assert catch_exit(Task.shutdown(task, :brutal_kill)) ==
-        {{:nodedown, node()}, {Task, :shutdown, [task, :brutal_kill]}}
+               {{:nodedown, node()}, {Task, :shutdown, [task, :brutal_kill]}}
     end
 
     test "returns exit on killing task after shutdown timeout" do
       caller = self()
 
-      task = Task.async(fn() ->
-        Process.flag(:trap_exit, true)
-        wait_and_send(caller, :ready)
-        Process.sleep(:infinity)
-      end)
+      task =
+        Task.async(fn ->
+          Process.flag(:trap_exit, true)
+          wait_and_send(caller, :ready)
+          Process.sleep(:infinity)
+        end)
 
       receive do: (:ready -> :ok)
       assert Task.shutdown(task, 1) == {:exit, :killed}


  describe "async_stream/2" do
    test "timeout" do
      assert catch_exit([:infinity] |> Task.async_stream(&sleep/1, [timeout: 0]) |> Enum.to_list) ==
             {:timeout, {Task.Supervised, :stream, [0]}}
      refute_received _
    end

    test "streams an enumerable with ordered: false" do
      opts = [max_concurrency: 1, ordered: false]
      assert 4..1 |> Task.async_stream(&sleep(&1 * 100), opts) |> Enum.to_list ==
             [ok: 400, ok: 300, ok: 200, ok: 100]

      opts = [max_concurrency: 4, ordered: false]
      assert 4..1 |> Task.async_stream(&sleep(&1 * 100), opts) |> Enum.to_list ==
             [ok: 100, ok: 200, ok: 300, ok: 400]
    end

    test "streams an enumerable with ordered: false, on_timeout: :kill_task" do
      opts = [max_concurrency: 4, ordered: false, on_timeout: :kill_task, timeout: 50]
      assert [100, 1, 100, 1] |> Task.async_stream(&sleep/1, opts) |> Enum.to_list() ==
             [ok: 1, ok: 1, exit: :timeout, exit: :timeout]
      refute_received _
    end

    test "streams an enumerable with infinite timeout" do
      [ok: :ok] = Task.async_stream([1], fn _ -> :ok end, timeout: :infinity) |> Enum.to_list
    end
  end

  for {desc, concurrency} <- ["==": 4, "<": 2, ">": 8] do
    describe "async_stream with max_concurrency #{desc} tasks" do
      @opts [max_concurrency: concurrency]

      test "streams an enumerable with fun" do
        assert 1..4 |> Task.async_stream(&sleep/1, @opts) |> Enum.to_list ==
               [ok: 1, ok: 2, ok: 3, ok: 4]
      end

      test "streams an enumerable with mfa" do
        assert 1..4 |> Task.async_stream(__MODULE__, :sleep, [], @opts) |> Enum.to_list ==
               [ok: 1, ok: 2, ok: 3, ok: 4]
      end

      test "streams an enumerable without leaking tasks" do
        assert 1..4 |> Task.async_stream(&sleep/1, @opts) |> Enum.to_list ==
               [ok: 1, ok: 2, ok: 3, ok: 4]
        refute_received _
      end

      test "streams an enumerable with slowest first" do
        Process.flag(:trap_exit, true)
        assert 4..1 |> Task.async_stream(&sleep/1, @opts) |> Enum.to_list ==
               [ok: 4, ok: 3, ok: 2, ok: 1]
      end

      test "streams an enumerable with exits" do
        Process.flag(:trap_exit, true)
        assert 1..4 |> Task.async_stream(&exit/1, @opts) |> Enum.to_list ==
               [exit: 1, exit: 2, exit: 3, exit: 4]
        refute_received {:EXIT, _, _}
      end

      test "shuts down unused tasks" do
        assert [0, :infinity, :infinity, :infinity] |> Task.async_stream(&sleep/1, @opts) |> Enum.take(1) ==
               [ok: 0]
        assert Process.info(self(), :links) == {:links, []}
      end

      test "shuts down unused tasks without leaking messages" do
        assert [0, :infinity, :infinity, :infinity] |> Task.async_stream(&sleep/1, @opts) |> Enum.take(1) ==
               [ok: 0]
        refute_received _
      end

      test "is zippable on success" do
        task = 1..4 |> Task.async_stream(&sleep/1, @opts) |> Stream.map(&elem(&1, 1))
        assert Enum.zip(task, task) ==
               [{1, 1}, {2, 2}, {3, 3}, {4, 4}]
      end

      test "is zippable on failure" do
        Process.flag(:trap_exit, true)
        task = 1..4 |> Task.async_stream(&exit/1, @opts) |> Stream.map(&elem(&1, 1))
        assert Enum.zip(task, task) ==
               [{1, 1}, {2, 2}, {3, 3}, {4, 4}]
      end

      test "is zippable with slowest first" do
        task = 4..1 |> Task.async_stream(&sleep/1, @opts) |> Stream.map(&elem(&1, 1))
        assert Enum.zip(task, task) ==
               [{4, 4}, {3, 3}, {2, 2}, {1, 1}]
      end

      test "with inner halt on success" do
        assert 1..8 |> Stream.take(4) |> Task.async_stream(&sleep/1, @opts) |> Enum.to_list ==
               [ok: 1, ok: 2, ok: 3, ok: 4]
      end

      test "with inner halt on failure" do
        Process.flag(:trap_exit, true)
        assert 1..8 |> Stream.take(4) |> Task.async_stream(&exit/1, @opts) |> Enum.to_list ==
               [exit: 1, exit: 2, exit: 3, exit: 4]
      end

      test "with inner halt and slowest first" do
        assert 8..1 |> Stream.take(4) |> Task.async_stream(&sleep/1, @opts) |> Enum.to_list ==
               [ok: 8, ok: 7, ok: 6, ok: 5]
      end

      test "with outer halt on success" do
        assert 1..8 |> Task.async_stream(&sleep/1, @opts) |> Enum.take(4) ==
               [ok: 1, ok: 2, ok: 3, ok: 4]
      end

      test "with outer halt on failure" do
        Process.flag(:trap_exit, true)
        assert 1..8 |> Task.async_stream(&exit/1, @opts) |> Enum.take(4) ==
               [exit: 1, exit: 2, exit: 3, exit: 4]
      end

      test "with outer halt and slowest first" do
        assert 8..1 |> Task.async_stream(&sleep/1, @opts) |> Enum.take(4) ==
               [ok: 8, ok: 7, ok: 6, ok: 5]
      end

      test "terminates inner effect" do
        stream =
          1..4
          |> Task.async_stream(&sleep/1, @opts)
          |> Stream.transform(fn -> :ok end,
                              fn x, acc -> {[x], acc} end,
                              fn _ -> Process.put(:stream_transform, true) end)

        Process.put(:stream_transform, false)
        assert Enum.to_list(stream) == [ok: 1, ok: 2, ok: 3, ok: 4]


      test "with :on_timeout set to :kill_task" do
        opts = Keyword.merge(@opts, on_timeout: :kill_task, timeout: 50)

        assert [100, 1, 100, 1]
               |> Task.async_stream(&sleep/1, opts)
               |> Enum.to_list() == [exit: :timeout, ok: 1, exit: :timeout, ok: 1]

        refute_received _
      end
    end
