Ayoush

Ayoush

Why the consumer constantly send demand upstream in one condition and not in others?

So in the example given here :-GenStage – gen_stage v0.14.0 that of A → B → C I implemented that and saw that the consumer is able to constantly send demands upstream and the numbers are getting printed. But when I created a normal consumer, producer

defmodule Asyncgenstage.Producer do
  use GenStage
  def start_link(_args)  do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  def init(args) do
    with {:ok, pid} <- :python.start([{:python_path, to_charlist("path")},{:python, 'python3'}]) do
      {:producer, pid}
    end
  end

  def handle_demand(demand, state) do
    result = call_python(state)
    {:noreply, result, state}
  end

  def call_python(pid) do
    call_MFA(pid)
  end

  defp call_MFA(pid) do
    module = :url_generator
    function = :main
    arguments = [10]
    :python.call(pid, module, function, arguments)
  end

end

defmodule Asyncgenstage.Consumer do
  use GenStage
  alias Asyncgenstage.Producer, as: Producer
  def start_link, do: start_link([])
  def start_link(_args) do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  def init(:ok) do
    {:consumer, :ok, subscribe_to: [{Producer, max_demand: 10, mid_demand: 1}]}
  end


  def handle_events(events, from, state) do
    IO.inspect(events)
    {:noreply, [], state}
  end
end

I observer the demand is not getting send continuously so I thought maybe the the function is not getting called so many times but when I introduced max_demand and min_demand in consumer then it start sending demand upstream constantly. Why is that, max_demand and min_demand are just opts right like by default also we have that value then why on explicitly declaring them is triggering the demand upstream constantly but without them its not.

Marked As Solved

LostKobrakai

LostKobrakai

You want to read the following part in the docs of a recent version of GenStage: GenStage — gen_stage v1.3.2

It explains how demand is handled. In your case the problem is likely the producer, which produces a fixed amount of events whenver handle_demend is triggered instead of producing as many events as there has been demand sent.

Your producer is expected to produce 5* events eventually if it received a demand of 5 in handle_demand. Consumers won’t demand more before their previous demand has been fulfilled to some degree (depending on min and max_demand setting).

[*] At least 5. GenStage can buffer some events exceeding demand to make e.g. producers working with batches of data simpler.

Where Next?

Popular in Questions Top

chokchit
** (DBConnection.ConnectionError) connection not available and request was dropped from queue after 2733ms. You can configure how long re...
New
9mm
I am constructing a JSON object (map) and I need to conditionally set a field. I’m trying to write proper elixir-way code… and I’m at a l...
New
senggen
Erlang/OTP 25 [erts-13.2.2] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] 15:22:35.803 [error] gen_event {lager_file_backend...
New
baxterw3b
Hi guys, i’m new in the Elixir world, and i have to say, that i love it! i’m having some problem to understand anonymous functions with ...
New
chrisalley
ExUnit now has describe blocks which is a welcome addition coming from RSpec. In the docs, it states that nested hierarchies of describe ...
New
mgjohns61585
Could someone help me? I’m making my first elixir program, number guessing game. I can’t figure out how to convert the user’s guess from ...
New
johnnyicon
Hi all, I’ve just started learning Elixir and Phoenix Framework, so please pardon my n00bness at this stage. I’m trying to use Postgres...
New
pmjoe
I have a relationship of love and hate with Elixir. Lots of things are just absolutely right, but there are some things that are kind of ...
New
Fl4m3Ph03n1x
About me? ( if you have nothing better to do than reading about some random guy in the internet :stuck_out_tongue: ) Hello all, this is ...
New
chensan
I have a User schema with a :from_id field set to type :string: defmodule TweetBot.Repo.Migrations.CreateUsers do use Ecto.Migration ...
New

Other popular topics Top

danschultzer
None of the current solutions worked well for me, so I went ahead and built a user management system from scratch. This project took far...
548 29603 241
New
sorentwo
Hello! tl;dr Announcing Oban, an Ecto based job processing library with a focus on reliability and historical observability. After spen...
985 43487 311
New
marius95
Hello everyone, I try to use an Javascript Event Handler in my root.html.leex file. Therefore I created a function in the app.js file: ...
New
Fl4m3Ph03n1x
About me? ( if you have nothing better to do than reading about some random guy in the internet :stuck_out_tongue: ) Hello all, this is ...
New
fireproofsocks
Forgive me if this is obvious, but how does one delete a database record WITHOUT selecting it first? Ecto.Repo — Ecto v3.14.0 has exampl...
New
jerry
Good day to you all. I have been struggling to get a query involving like and ilike to work. Can anyone assist me on this, please? pro...
New
josevalim
Hi everyone, One of the features added to Elixir early on to help integration with Erlang code was the idea of overridable function defi...
New
alice
Hey, Just curious what are the main benefits of Elixir compared to Clojure? When is Elixir more useful than Clojure and vice versa? Th...
New
RisingFromAshes
I’ve read in another post that it may be possible with a router helper - but I couldn’t find an appropriate one, and tbh, I’m still just ...
New
JakeBecker
TL;DR: I’ve just released an implementation of Microsoft’s IDE-independent Language Server Protocol for Elixir. It adds language support ...
1144 54120 245
New

Latest on Elixir Forum

Elixir Forum

We're in Beta

About us Mission Statement