Phoenix.Channel + GenStage

consumer_channel.ex

defmodule MyApp.Channel do
  alias Experimental.GenStage
  use GenStage
  use MyApp.Web, :channel
  def init(socket) do
    {:consumer, socket, subscribe_to: [MyApp.BroadcastStage]}
  end
  def handle_events(events, _from, socket) do
    # get events pushed directly to each channel connection
    Enum.each events, &Phoenix.Channel.push(socket, "message", &1)
    {:noreply, [], socket}
  end
  def join(topic, params, socket) do
    # call start_link when you want to accept the connection
    GenStage.start_link(__MODULE__, socket)
    {:ok, socket}
  end
end

I only tried this for consumers. Producer stages need much more work since handle_in, etc. don't support the {reply, events, state} return.