Packages

smart_city_test

0.2.5
3.0.0 2.4.0 2.3.0 2.2.8 2.2.7 2.2.5 2.2.4 2.2.3 2.2.2 2.2.1 2.2.0 2.1.4 2.1.3 2.1.2 2.1.1 2.1.0 2.0.5 2.0.4 2.0.3 2.0.2 2.0.1 2.0.0 1.0.0 0.10.1 0.10.0 0.9.3 0.9.2 0.9.1 0.9.0 0.8.1 0.8.0 0.7.0 0.6.0 0.5.4 0.5.3 0.5.2 0.5.1 0.5.0 0.4.0 0.3.5 0.3.4 0.3.3 0.3.2 0.3.0 0.2.7 0.2.6 0.2.5

A library that contains test utilites used in the Smart City project

HexDocs

Current section

Files

Jump to
Readme
47 Versions
3 Dependencies
0 Dependants
Files
Activity
smart_city_test lib smart_city kafka_helper.ex
Raw

lib/smart_city/kafka_helper.ex 

defmodule SmartCity.KafkaHelper do

  @moduledoc false



  def stream_messages(messages, topic, chunk_size \\ 5) when is_list(messages) do

    messages

    |> Stream.cycle()

    |> Stream.chunk_every(chunk_size)

    |> Stream.map(&send_to_kafka(&1, topic))

    |> Stream.run()

  end



  def send_to_kafka(msg, topic) when is_list(msg) do

    msg

    |> Enum.map(&send_to_kafka(&1, topic))

    |> Enum.find(:ok, &(&1 != :ok))

  end



  def send_to_kafka(msg, topic) do

    json_msg = apply(Jason, :encode!, [msg])

    apply(Kaffe.Producer, :produce_sync, [topic, [{"the_key", json_msg}]])

  end

end