alvises

GenStage and twitter stream

Hi guys,

I’m playing a bit with GenStage, after reading the blog post on elixir with a full case description, and watching the youtube Jose’s speech.
I’m playing with it using twitter streams. The producer reads the twitter streams, receiving all the tweets linked to the hashtag #rio2016, and pass them to the consumers.

in the gist there are app, producer, consumer and log output

gist.github.com

https://gist.github.com/alvises/656cacf7b1962927b34f2708e52ebc6f

app.ex

defmodule Twitter do
	
	def start(_type, _args) do
		import Supervisor.Spec

		workers = [
			worker(Twitter.Producer, ["#Rio2016"]),
			worker(Twitter.Consumer, [], id: 1),
			worker(Twitter.Consumer, [], id: 2),
			worker(Twitter.Consumer, [], id: 3)

This file has been truncated. show original

consumer.ex

alias Experimental.GenStage

defmodule Twitter.Consumer do
	use GenStage

	def start_link do
		GenStage.start_link(__MODULE__, [])
	end

	def init(tweets) do

This file has been truncated. show original

output

[#PID<0.177.0>,
 "RT @BBCSport: It was a sensational #Rio2016... Welcome home @TeamGB 👏 https://t.co/IerPuEso63"]
[#PID<0.177.0>,
 "RT @WhiteHouse: ICYMI: Go behind the scenes with the the U.S. delegation at the #Rio2016 Closing Ceremony: https://t.co/G6iq5occa4 https://…"]
[#PID<0.177.0>,
 "RT @sportskh: #Rio2016 #리우 #올림픽 #수영 호주 캠벨, 탈장 상태로 올림픽 출전\n\n자세한 기사는\nhttps://t.co/8ZR6cBEfng https://t.co/58wXQ1rv3Z"]
[#PID<0.177.0>,
 "RT @SocialSA_: Van Niekerk will receive an extra R150k, Caster an extra R100k‚ Silver winners receive extra R70k , and for Bronze R50k extr…"]
[#PID<0.177.0>,
 "RT @Ajay_Chandrakar: Congratulations #PVSindhu for Getting First Ever #Silver in Badminton @ #Rio2016"]

This file has been truncated. show original

There are more than three files. show original

I’ve started 3 consumers. Once a consumer receives a bunch of tweets from the producer, I put it to sleep for 10s to simulate the processing. The problem is that the other 2 consumers don’t receive anything…

if I change the stream to a local big file, all the 3 consumers receive events…

Does anyone know why ?

Thanks

Alvise

20 comments

#twitter

31 4450 20

2018-05-15 22:29:02 UTC

Most Liked

mario

When my GenStage consumer(s) are too slow to handle the amount of incoming data (Tweets) the producer will stop ack the HTTP/TCP connection packages. The Twitter streaming endpoint will stop sending more data through the connection.

This is what whe want and follows the same pattern GenStage and Flow follow to consume data.

Without back-pressure, my producer would have to buffer the incoming Tweets or it’s internal message box would grow indefinitly. In both cases this would result in taking more and more memory, crashing the VM in the long term.

Post #11

mario

I wrote a Twitter library from scratch (at that time, it was my first project to get startet with Elixir). It uses HTTPoison (wrapper for :hackney).

A few weeks ago, I’ve updated the streaming part of the lib to use GenStage.

You can have a look at the implementation here: https://github.com/almightycouch/twittex/blob/master/lib/twittex/client/stream.ex

Basically, the Twittex.Client.Stream receives incoming HTTP chunks from :hackney applying back-pressure at the TCP level with :hackney.stream_next/1.

The next thing I want to try out is using Flow and Flow.Window to implement features such as peak detection.

Post #9

josevalim

Creator of Elixir

It is a bug in ExTwitter. The stream is discarding messages from the process inbox, which is a big no-no. This means that the subscriptions for the second and third stages are never consumed:

ExTwitter should use a reference and make sure it only consumes messages it knows about. This must be reported as a bug.

Ideally ExTwitter would provide a GenStage producer, to avoid hijacking the process inbox, which won’t work with a GenStage since both GenStage and ExTwitter are trying to use the same process inbox. You can check the issues tracker so see if someone already requested this feature.

Post #7

MarioFlach

Hi @maz, no problem at all, you helped me found an annoying bug here :

@api_key Application.get_env(:twittex, :consumer_key)
@api_secret Application.get_env(:twittex, :consumer_secret)

Basically, @consumer_key and @consumer_secret where defined at compile time in the Twittex.Client.Base module. Updating those configuration keys did not force the module to recompile which I think is why you encountered you authentication problems (the first time you compiled :twittex, those keys where compiled into the application even when you updated your config).

You could try to run mix deps.clean twittex and recompile your application (or run iex -S mix directly), this should fix the problem. Otherwise, I just published version 0.3.5 which fixes the issue and loads :consumer_key and :consumer_secret at runtime.

Post #19

benwilson512

Author of Craft GraphQL APIs in Elixir with Absinthe

You haven’t overridden the default demand quantities. GenStage operates in batches, and the min / max batch sizes are 500 / 1000 by default, so unless you have more than that it isn’t gonna use more than one worker.

This is a good example of https://github.com/elixir-lang/gen_stage/issues/72

Post #2