flow_example.ex

def FlowExample do
  def process(tickets_batch) do
      customer_flow = create_customer_flow(tickets_batch)
      messages_flow = create_messages_flow(tickets_batch)

      Flow.bounded_join(:inner,
                        customer_flow,
                        messages_flow,
                        &(&1.number),
                        &(&1.number),
                        &TicketProcessor.update_ticket_flag/2) |> Flow.run

    end

    defp create_customer_flow(tickets_batch) do
      tickets_batch
      |> Flow.from_enumerable
      |> Flow.partition(max_demand: 50, stages: 10)
      |> Flow.map(&CustomerCreator.import_customer/1)
    end

    defp create_messages_flow(tickets_batch) do
      tickets_batch
      |> Flow.from_enumerable
      |> Flow.partition(max_demand: 50, stages: 10)
      |> Flow.map(&TicketProcessor.import_messages/1)
    end
  end
end