refactor(DetailedStopGroup): replace async_stream

lib/detailed_stop_group.ex

@@ -38,26 +38,25 @@ defmodule DetailedStopGroup do
     mode
     |> Route.types_for_mode()
     |> @routes_repo.by_type()
-    |> Task.async_stream(&{&1, @stops_repo.by_route(&1.id, 1)})
-    |> Enum.map(fn {:ok, stops} -> stops end)
+    |> Task.async_stream(&{&1, @stops_repo.by_route(&1.id, 1)}, max_concurrency: 10, on_timeout: :kill_task)
+    |> Stream.map(fn {:ok, stops} -> stops end)
   end
 
   @spec from_grouped_stops([grouped_stops]) :: [
           DetailedStopGroup.t()
         ]
   defp from_grouped_stops(grouped_stops) do
     grouped_stops
-    |> Task.async_stream(&build_featured_stops(&1))
-    |> Enum.map(fn {:ok, featured_stops} -> featured_stops end)
+    |> Stream.map(&build_featured_stops/1)
   end
 
   @spec build_featured_stops(grouped_stops) :: DetailedStopGroup.t()
   defp build_featured_stops({route, stops}) do
     featured_stops =
       stops
-      |> Enum.sort_by(& &1.name)
-      |> Task.async_stream(&build_featured_stop(route, &1))
-      |> Enum.map(fn {:ok, featured_stop} -> featured_stop end)
+      |> Task.async_stream(&build_featured_stop(route, &1), max_concurrency: 10, on_timeout: :kill_task)
+      |> Stream.map(fn {:ok, featured_stop} -> featured_stop end)
+      |> Enum.sort_by(& &1.stop.name)
 
     {route, featured_stops}
   end

lib/dotcom_web/controllers/stop_controller.ex

@@ -256,8 +256,7 @@ defmodule DotcomWeb.StopController do
   @spec get_stop_info :: {DetailedStopGroup.t(), [DetailedStopGroup.t()]}
   defp get_stop_info do
     [:subway, :commuter_rail, :ferry]
-    |> Task.async_stream(&DetailedStopGroup.from_mode/1)
-    |> Enum.flat_map(fn {:ok, stops} -> stops end)
+    |> Stream.flat_map(&DetailedStopGroup.from_mode/1)
     |> separate_mattapan()
   end
 
@@ -267,7 +266,7 @@ defmodule DotcomWeb.StopController do
   defp separate_mattapan(stop_info) do
     case Enum.find(stop_info, fn {route, _stops} -> route.id == "Mattapan" end) do
       nil -> {nil, stop_info}
-      mattapan -> {mattapan, List.delete(stop_info, mattapan)}
+      mattapan -> {mattapan, Stream.reject(stop_info, & &1 == mattapan)}
     end
   end