 
   defp stream_close(waiting, monitor_pid, monitor_ref, timeout) do
     for {_, {pid, _}} <- waiting do
-      Process.unlink(pid)
+      pid && Process.unlink(pid)
     end
     send(monitor_pid, {:DOWN, monitor_ref})
     receive do

      {:DOWN, {^monitor_ref, position}, reason} ->
        waiting =
          case waiting do
            # We update the entry only if it is running.
            # If it is ok or removed, we are done.
            %{^position => {pid, :running}} -> Map.put(waiting, position, {pid, {:exit, reason}})
            %{} -> waiting
          end
        stream_deliver({:cont, acc}, max + 1, spawned, delivered, waiting, next,
                       reducer, mfa, spawn, monitor_pid, monitor_ref, timeout)

  defp stream_deliver({:cont, acc}, max, spawned, delivered, waiting, next,
                      reducer, mfa, spawn, monitor_pid, monitor_ref, timeout) do
    case waiting do
      %{^delivered => {nil, reply}} ->
        try do
          reducer.(reply, acc)
        catch
