I have a function that can operate on a single struct of a specific type, a list of those structs, or a stream of those structs:

def update_data(%MyStruct{} = my_struct), 
  do: %MyStruct{my_struct | data: "Here's the new data!"}
  
def update_data(%Stream{} = stream),
  do: Stream.map(stream, &update_data/1)

def update_data(structs) when is_list(structs),
  do: Enum.map(structs, &update_data/1)

def update_data({:ok, value}),
  do: {:ok, update_data(value)}

def update_data(value),
  do: IO.inspect(value, label: "update_data fell through")

This is working great under almost all conditions. The only exception is when the last step in the pipe before update_data/1 is Stream.flat_map/2. I have discovered this is because the %Stream{} = stream pattern match is failing, because the value returned from Stream.flat_map/2 isn’t a stream:

Interactive Elixir (1.13.3) - press Ctrl+C to exit (type h() ENTER for help)
iex(1)> [1, 2, 3] |> Stream.map(& &1)
#Stream<[enum: [1, 2, 3], funs: [#Function<47.58486609/1 in Stream.map/2>]]>
iex(2)> [1, 2, 3] |> Stream.each(&IO.puts/1)
#Stream<[enum: [1, 2, 3], funs: [#Function<38.58486609/1 in Stream.each/2>]]>
iex(3)> [1, 2, 3] |> Stream.flat_map(& &1)  
#Function<59.58486609/2 in Stream.transform/3>

I guess that means Stream.transform/3 doesn’t return a new stream, either, but a function as well.

I know I can work around this in a couple of different ways, but none of them seem ideal. I don’t want to make the caller responsible for making sure the thing being piped to update_data/1 is an actual Stream. I also don’t want to assume any given Function that shows up is pipe-able to Stream.map/2.

What’s the best way to get update_data/1 to behave the way I want it to?