I’m fairly new to Elixir, and so far I’m finding it very exciting. So many of my concurrency problems seem to be solved in a simple, elegant way. Now i’ve learnt the basics, I’m trying to write some applications and am having trouble working out the right patterns.

First I was using Wabbit as a Genstage producer (consuming from a rabbitmq queue). I was doing some processing on the messages and then inserting into a DB. It worked great locally, around ~8000/s. When I moved it to a production environment, after excessive logging, I worked out that it was choking up at the Sink end (backpressure was working!) because of latency of writes. Locally, i would get say, 500 events to the consumer at a time, but in production i would get 1 at a time, which would make for higher latency on the write per message.

My solution was to create a stage which buffered upto x events, or for x ms (whichever came first). To be more concrete, here is the code.

# This will handle the acks back to rabbitmq
defmodule RabbitAckker do
  use GenStage
  require Logger

  def start_link() do
    GenStage.start_link(__MODULE__, :ok)
  end

  def init(:ok) do
    Process.send_after(self(), {:flush}, 500) # first message is fairly soon after startup
    {:producer_consumer, []}
  end

  # handle events coming from rabbit
  def handle_events(events, _from, state) do
    event_buffer = state
    new_event_buffer = event_buffer ++ events
    messages_before_purge = 400

    if (Enum.count(new_event_buffer) >= messages_before_purge) do
      for {event, meta} <- new_event_buffer do
        :ok = ack(meta.channel, meta.delivery_tag)
      end

      # pass all events down to the next stage
      {:noreply, new_event_buffer, []}
    else
      # don't pass any events down, just store them in the state
      {:noreply, [], new_event_buffer}
    end
  end

  # when this flush message comes, ack the events and pass them down.
  # TODO: only flush when we haven't acked anything in flush_time ms
  def handle_info({:flush}, state) do
    flush_time = 2_000 # ms
    event_buffer = state

    event_count = Enum.count(state)
    if event_count > 0 do
      Logger.debug("ackker: timer flush called for #{event_count} events")
    end
    # ack all the events we will flush
    for {event, meta} <- event_buffer do
      :ok = ack(meta.channel, meta.delivery_tag)
    end
    
    # call this again in another flush_time ms
    Process.send_after(self(), {:flush}, flush_time)
    # move event_buffer to events, and reset the state
    {:noreply, event_buffer, []}
  end

  defp ack(channel, delivery_tag) do
    try do
      Wabbit.Basic.ack(channel, delivery_tag)
    catch
      _, _ ->
        :ok
    end
  end
end

This works great! I realise that I have potential for dataloss if the process is killed before flush_time but you have that problem with any sort of buffering.

My first question is, is there a better way to do this?

Now on to my next problem!

In addition to writing to a datastore, I would like to spit out the messages to multiple websockets. Sometimes the consumers on the end of the websocket are slow. Each of them have different performance characteristics. How can I drop messages to them if they aren’t keeping up? I am trying to avoid an unbounded buffer. I am also trying to write to the datastore at the same time, but i want the datastore to get every message and apply backpressure (via demand) as it is in the above example.

Any suggestions on a good pattern for this?

Thanks for taking the time to read through my whole question!