Packages

flowex

0.3.0
0.5.4 0.5.3 0.5.2 0.5.1 0.5.0 0.4.0 0.3.0 0.2.0 0.1.2 0.1.1 0.1.0

Flow-Based Programming with Elixir GenStage.

HexDocs

Current section

11 Versions

Jump to
Readme
11 Versions
1 Dependency
Dependants
Files
Activity
Hide whitespace

Compare versions

7 files changed
+102 additions
-35 deletions
  @@ -2,8 +2,8 @@
2 2 [![Build Status](https://travis-ci.org/antonmi/flowex.svg?branch=master)](https://travis-ci.org/antonmi/flowex)
3 3 [![Hex.pm](https://img.shields.io/hexpm/v/flowex.svg?style=flat-square)](https://hex.pm/packages/flowex)
4 4
5 - ## Railway Flow-Based Programming with Elixir GenStage.
6 - #### Flowex is a set of abstractions build on top Elixir GenStage which allows writing program with [Flow-Based Programming](https://en.wikipedia.org/wiki/Flow-based_programming) paradigm.
5 + ## Railway Flow-Based Programming.
6 + #### Flowex is a set of abstractions built on top Elixir GenStage which allows writing program with [Flow-Based Programming](https://en.wikipedia.org/wiki/Flow-based_programming) paradigm.
7 7 I would say it is a mix of FBP and so-called [Railway Oriented Programming (ROP)](http://fsharpforfunandprofit.com/rop/) approach.
8 8
9 9 Flowex DSL allows you to easily create "pipelines" of Elixir GenStages.
  @@ -19,7 +19,8 @@ Flowex DSL allows you to easily create "pipelines" of Elixir GenStages.
19 19 - [Error handling](#error-handling)
20 20 - [Synchronous and asynchronous calls](#synchronous-and-asynchronous-calls)
21 21 - [Bottlenecks](#bottlenecks)
22 - - [Module pipelines](#module-pipelines)
22 + - [Module pipes](#module-pipes)
23 + - [Starting strategies](#starting-strategies)
23 24 - [Contributing](#contributing)
24 25
25 26 ## Installation
  @@ -128,7 +129,7 @@ We also renamed the module to `FunPipeline` because we are going to create "Flow
128 129 `Flowex.Pipeline` extend our module, so we have:
129 130 - `pipe` macro to define which function evaluation should be placed into separate GenStage;
130 131 - `error_pipe` macro to define function which will be called if error occurs;
131 - - `start` and `stop` functions to create and destroy pipelines;
132 + - `start`, `supervised_start` and `stop` functions to create and destroy pipelines;
132 133 - `call` function to run pipeline computations synchronously.
133 134 - `cast` function to run pipeline computations asynchronously.
134 135
  @@ -144,10 +145,10 @@ pipeline = FunPipeline.start(opts)
144 145 sup_pid: #PID<0.136.0>}
145 146 ```
146 147 What happened:
147 - - Three GenStages were started - one for each of the function in pipeline. Each of GenStages is `:producer_consumer`;
148 - - One additional GenStage for error processing is started (it is also `:producer_consumer`);
149 - - Runs 'producer' and 'consumer' GenStages for input and output;
150 - - All the components are placed under Supervisor.
148 + - Three GenStages have been started - one for each of the function in pipeline. Each of GenStages is `:producer_consumer`;
149 + - One additional GenStage for error processing has been started (it is also `:producer_consumer`);
150 + - 'producer' and 'consumer' GenStages for input and output have been added;
151 + - All the components have been placed under Supervisor.
151 152
152 153 The next picture shows what the 'pipeline' is.
153 154 ![alt text](figures/fun_pipeline.png "FunPipeline")
  @@ -159,6 +160,8 @@ The `start` function returns a `%Flowex.Pipeline{}` struct with the following fi
159 160 - sup_pid - pid of the pipeline supervisor
160 161
161 162 Note, we have passed options to `start` function. This options will be passed to each function of the pipeline as a second argument.
163 + There is `supervised_start` function which allows to place pipeline's under external supervisor.
164 + See details in [Starting strategies](#starting-strategies) section.
162 165
163 166 ## Run the pipeline
164 167 One can run calculations in pipeline synchronously and asynchronously:
  @@ -282,7 +285,7 @@ And the pipeline will look like on the figure below:
282 285 ![alt text](figures/complex_pipeline.png "Group of clients")
283 286
284 287
285 - ## Module pipelines
288 + ## Module pipes
286 289 One can create reusable 'pipe' - module which implements init and call functions.
287 290 ```elixir
288 291 defmodule ModulePipeline do
  @@ -344,6 +347,62 @@ end
344 347
345 348 Of course, one can combine module and functional 'pipes'!
346 349
350 + ## Starting strategies
351 + Using `start/1` function one can start pipelines in any process. Pipelines will be alive while the process is alive.
352 + The `supervised_start` function accepts supervisor `pid` as the first argument and `opts` as the second argument.
353 + And starts pipeline's supervisor under predefined supervisor process.
354 +
355 + In general there are three ways to start pipelines in your project:
356 +
357 + 1. Start pipelines in arbitrary supervised process:
358 + ```elixir
359 + defmodule PipelineGenServer do
360 + use GenServer
361 +
362 + def init(_opts) do
363 + pipeline_one = PipelineOne.start
364 + pipeline_two = PipelineTwo.start
365 +
366 + {:ok, %{pipeline_one: pipeline_one, pipeline_two: pipeline_two}}
367 + end
368 + end
369 + ```
370 + You can also store pipeline structure in Agent or Application environment.
371 +
372 + 2. Start one pipeline per application. In that case pipeline supervisor will be the main supervisor in the application:
373 + ```elixir
374 + defmodule OnePipelinePerApp do
375 + use Application
376 +
377 + def start(_type, _opts) do
378 + pipeline = PipelineOne.start
379 + Application.put_env(:start_flowex, :pipeline, pipeline)
380 + {:ok, pipeline.sup_pid}
381 + end
382 + end
383 + ```
384 +
385 + 3. Start several pipelines inside one application using `supervised_start` function. In that case pipeline supervisors will be placed under application supervisor:
386 + ```elixir
387 + defmodule TwoPipelinesPerApp do
388 + use Application
389 +
390 + def start(_type, _opts) do
391 + {:ok, supervisor_pid} = Supervisor.start_link([], strategy: :one_for_one, name: :multi_flowex_sup)
392 +
393 + pipeline_one = PipelineOne.supervised_start(supervisor_pid)
394 + pipeline_two = PipelineTwo.supervised_start(supervisor_pid)
395 +
396 + Application.put_env(:start_flowex, :pipeline_one, pipeline_one)
397 + Application.put_env(:start_flowex, :pipeline_two, pipeline_two)
398 +
399 + {:ok,supervisor_pid}
400 + end
401 + end
402 + ```
403 +
404 + You can find the examples in ['Start-Flowex'](https://github.com/antonmi/Start-Flowex) project
405 +
347 406 ## Contributing
348 407 #### Contributions are welcome and appreciated!
  @@ -17,4 +17,4 @@
17 17 {<<"name">>,<<"gen_stage">>},
18 18 {<<"optional">>,false},
19 19 {<<"requirement">>,<<"0.11.0">>}]]}.
20 - {<<"version">>,<<"0.2.0">>}.
20 + {<<"version">>,<<"0.3.0">>}.
  @@ -27,6 +27,10 @@ defmodule Flowex.Pipeline do
27 27 Flowex.PipelineBuilder.start(__MODULE__, opts)
28 28 end
29 29
30 + def supervised_start(pid, opts \\ %{}) do
31 + Flowex.PipelineBuilder.supervised_start(__MODULE__, pid, opts)
32 + end
33 +
30 34 def stop(%Flowex.Pipeline{sup_pid: sup_pid}) do
31 35 Flowex.PipelineBuilder.stop(sup_pid)
32 36 end
  @@ -44,32 +48,28 @@ defmodule Flowex.Pipeline do
44 48
45 49 def call(%Flowex.Pipeline{in_name: in_name, out_name: out_name} = pipeline, struct = %__MODULE__{}) do
46 50 pid = self()
47 - out_pid = link_to_consumer(out_name)
51 + ref = Process.monitor(out_name)
48 52 ip = %Flowex.IP{struct: struct, requester: pid}
53 +
49 54 GenServer.cast(out_name, {in_name, ip})
55 +
50 56 receive do
51 57 %Flowex.IP{requester: ^pid} = ip ->
58 + Process.demonitor(ref)
52 59 ip.struct
53 - {:EXIT, ^out_pid, reason} ->
60 + {:DOWN, ^ref, _, _, reason} ->
54 61 raise Flowex.PipelineError, pipeline: pipeline, message: reason
55 62 smth ->
56 63 reason = "Expected %Flowex.IP{}, received #{inspect smth}"
57 64 raise Flowex.PipelineError, pipeline: pipeline, message: reason
58 65 end
66 +
59 67 end
60 68
61 69 def cast(%Flowex.Pipeline{in_name: in_name, out_name: out_name} = pipeline, struct = %__MODULE__{}) do
62 70 ip = %Flowex.IP{struct: struct, requester: false}
63 71 GenServer.cast(out_name, {in_name, ip})
64 72 end
65 -
66 - defp link_to_consumer(out_name) do
67 - out_pid = Process.whereis(out_name)
68 -
69 - Process.link(out_pid)
70 - Process.flag(:trap_exit, true)
71 - out_pid
72 - end
73 73 end
74 74 end
75 75 end
  @@ -3,6 +3,24 @@ defmodule Flowex.PipelineBuilder do
3 3
4 4 def start(pipeline_module, opts) do
5 5 {:ok, sup_pid} = Flowex.Supervisor.start_link(pipeline_module)
6 + do_start(sup_pid, pipeline_module, opts)
7 + end
8 +
9 + def supervised_start(pipeline_module, pid, opts) do
10 + sup_spec = supervisor(Flowex.Supervisor, [pipeline_module], [id: "Flowex.Supervisor_#{inspect make_ref()}", restart: :permanent])
11 + {:ok, sup_pid} = Supervisor.start_child(pid, sup_spec)
12 + do_start(sup_pid, pipeline_module, opts)
13 + end
14 +
15 + def stop(sup_pid) do
16 + Supervisor.which_children(sup_pid)
17 + |> Enum.each(fn({id, _pid, :worker, [_]}) ->
18 + Supervisor.terminate_child(sup_pid, id)
19 + end)
20 + Supervisor.stop(sup_pid)
21 + end
22 +
23 + defp do_start(sup_pid, pipeline_module, opts) do
6 24 [{producer_name, _in_producer, :worker, [Flowex.Producer]}] = Supervisor.which_children(sup_pid)
7 25
8 26 last_names = (pipeline_module.pipes() ++ [pipeline_module.error_pipe])
  @@ -23,14 +41,6 @@ defmodule Flowex.PipelineBuilder do
23 41 %Flowex.Pipeline{module: pipeline_module, in_name: producer_name, out_name: consumer_name, sup_pid: sup_pid}
24 42 end
25 43
26 - def stop(sup_pid) do
27 - Supervisor.which_children(sup_pid)
28 - |> Enum.each(fn({id, _pid, :worker, [_]}) ->
29 - Supervisor.terminate_child(sup_pid, id)
30 - end)
31 - Supervisor.stop(sup_pid)
32 - end
33 -
34 44 defp init_function_pipe(sup_pid, {type, pipeline_module, function, opts}, prev_pids) do
35 45 name = String.to_atom("Flowex_#{pipeline_module}.#{function}_#{inspect make_ref()}")
36 46 worker_spec = worker(Flowex.Stage,
  @@ -33,7 +33,7 @@ defmodule Flowex.Stage do
33 33 %{ip | struct: apply(module, function, [ip.struct, opts])}
34 34 rescue
35 35 error ->
36 - %{ip | error: %Flowex.PipeError{message: error.message, pipe: {module, function, opts}, struct: ip.struct}}
36 + %{ip | error: %Flowex.PipeError{message: Exception.message(error), pipe: {module, function, opts}, struct: ip.struct}}
37 37 end
38 38 end
39 39 end
Loading more files…