scripbox / flume Goto Github PK
View Code? Open in Web Editor NEWA blazing fast job processing system backed by GenStage & Redis.
A blazing fast job processing system backed by GenStage & Redis.
One less external dependency.
Thank you for this amazing library.
Is there a way to cancel a scheduled job?
Thanks
flume.producer.init = system_time
)flume.producer.events.fetch = 10
)flume.producer.events.fetch.stop
):pipeline_name
- name of the pipeline for which producer is running:queue_name
- name of the queue in the redis
:rate_limit_count
- no. of events that can be processed by the consumer:rate_limit_scale
- duration for which no. of events as per rate_limit_count
can be processed:rate_limit_key
- unique key for which rate limit is applied:max_demand
- Maximum number of jobs to pull from the queueflume.producer_consumer.init = system_time
)flume.producer_consumer.events.received = 10
)flume.consumer.init = system_time
)cc @nitinstp23
This API can be used by apps for instrumentation purpose.
Currently flume defaults to Flume.DefaultLogger
. And we make clients define their own logger module Flume.Logger
. For some reasons we added this. We don't need it, we just need to remove these two module and use Logger
library.
An invalid job is accepted. It fails and then gets retried for the configured number of times.
An invalid job can be defined to have one or more of the following set incorrectly
A job in the aforementioned case is bound to fail and eventually move to the dead queue.
Therefore, we can return an error before enqueing in such cases by checking the following:
The asynchronous calls for pause/resume fail with
iex(1)> Flume.pause(:default_pipeline, async: true)
** (FunctionClauseError) no function clause matching in Flume.Pipeline.Event.Producer.pause/3
The following arguments were given to Flume.Pipeline.Event.Producer.pause/3:
# 1
"default_pipeline"
# 2
true
# 3
5000
Attempted function clauses (showing 1 out of 1):
def pause(pipeline_name, false = _async, timeout)
(flume) lib/flume/pipeline/event/producer.ex:29: Flume.Pipeline.Event.Producer.pause/3
The function definitions at
The Flume configuration works with a single Redis server.
However, it would not work in a setup with Redis Sentinels.
Extend the configuration to accept Sentinel options.
The underlying Redis library supports and manages the switching to a master server after a fail-over.
Currently, the Flume.pause
and Flume.resume
issues synchronous calls to the Producer GenStage to pause/resume pipeline.
Package should also support asynchronous calls to allow for cases where GenStage calls timeout and the operation fails.
On a Flume.pause("pipeline-name", _temporary = false)
call, the key in Redis store is set even in the case when Producer GenStage pause call fails.
On a Flume.resume("pipeline-name", _temporary = false)
call, the key in Redis store is deleted even in the case when Producer GenStage resume call fails.
This leads to an inconsistent pipeline state.
We should set/delete the Redis key on success of Producer GenStage pause/resume.
One advantage is that we can avoid using redis in test environment.
Usage will be similar to this
import Flume.Mock
describe "enqueue/4" do
test "mock works" do
with_flume_mock do
Flume.enqueue(:test, List, :last, [[1]])
assert_receive %{
queue: :test,
worker: List,
function_name: :last,
args: [[1]]
}
end
end
end
On mocking a Flume enqueue in test env. we do not serialize the args. This is inconsistent with how the worker module receives the args on a dequeue from the queue.
Flume.enqueue(:test, WorkerModule, :do_it, [%{a: 1, b: 2}])
with_flume_mock do
Flume.enqueue(:test, WorkerModule, :do_it, [%{a: 1, b: 2}])
assert_receive %{
queue: :test,
worker: WorkerModule,
function_name: :do_it,
args: [%{a: 1, b: 2}]
}
end
However, in an environment without mocks the job is serialized(Jason.encode!/1
) before we enqueue it.
This may differ from a possible expectation of a worker to receive serialized arguments.
The worker module executes the following equivalent of the operation
Flume.enqueue(:test, WorkerModule, :do_it, [%{"a" => 1, "b" => 2}])
So, the assertion should actually be against a serialized-deserialized version of the arguments.
with_flume_mock do
Flume.enqueue(:test, WorkerModule, :do_it, [%{a: 1, b: 2}])
assert_receive %{
queue: :test,
worker: WorkerModule,
function_name: :do_it,
args: [%{"a" => 1, "b" => 2}]
}
end
Update the Mocked API to serialize and deserialize the arguments.
args = Jason.encode!(args) |> Jason.decode!()
# 10 seconds
schedule_time = 10_000
Flume.enqueue_in(:queue_name, schedule_time, MyApp.FancyWorker, [arg_1, arg_2])
Here the schedule_time
is shown to be in milliseconds. It should actually be unix time in seconds.
Also, Flume.enqueue_in/4
, Flume.enqueue_in/5
and Flume.enqueue_in/6
accept argument time_in_seconds
. It is confusing. A better name for this argument would be unix_time_in_seconds
.
The current approach require us to define config options in config.exs
(or env specific files).
Moving config options to Flume.Supervisor
will have these benefits -
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.