Packages
electric
1.4.4
1.7.8
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.10
1.6.9
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.16
1.4.16-beta-1
1.4.15
1.4.14
1.4.13
1.4.12
1.4.11
1.4.10
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.4
1.3.3
1.3.2
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.14
1.1.13
1.1.12
1.1.11
1.1.10
1.1.9
1.1.8
1.1.7
1.1.6
retired
1.1.5
retired
1.1.4
retired
1.1.3
retired
1.1.2
1.1.1
1.1.0
1.0.24
1.0.23
1.0.22
1.0.21
1.0.20
1.0.19
1.0.18
1.0.17
1.0.15
1.0.13
1.0.12
1.0.11
1.0.10
1.0.9
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
1.0.0-beta.23
1.0.0-beta.22
1.0.0-beta.20
1.0.0-beta.19
1.0.0-beta.18
1.0.0-beta.17
1.0.0-beta.16
1.0.0-beta.15
1.0.0-beta.14
1.0.0-beta.13
1.0.0-beta.12
1.0.0-beta.11
1.0.0-beta.10
1.0.0-beta.9
1.0.0-beta.8
1.0.0-beta.7
1.0.0-beta.6
1.0.0-beta.5
1.0.0-beta.4
1.0.0-beta.3
1.0.0-beta.2
1.0.0-beta.1
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.3
0.6.2
0.6.1
0.5.2
0.4.4
Postgres sync engine. Sync little subsets of your Postgres data into local apps and services.
Current section
Files
Jump to
Current section
Files
lib/electric/concurrent_stream.ex
defmodule Electric.ConcurrentStream do
@default_poll_time 10
@doc """
Allows concurrent reading while writing of a stream.
There can be mutiple reading processes however there must be only one writing process.
The writing process must append an end marker to the end of the stream when it has finished
to signal to the reading processes that the stream has ended.
If a read process runs out of data to read before the end marker has been written
it waits the `poll_time_in_ms` for more data to be written, then resumes the stream
with the `stream_fun`.
"""
def stream_to_end(opts) do
excluded_start_key = Keyword.fetch!(opts, :excluded_start_key)
end_marker_key = Keyword.fetch!(opts, :end_marker_key)
stream_fun = Keyword.fetch!(opts, :stream_fun)
stream_fun.(excluded_start_key, end_marker_key)
|> continue_if_not_ended(excluded_start_key, opts)
end
defp continue_if_not_ended(stream, latest_key, opts) do
end_marker_key = Keyword.fetch!(opts, :end_marker_key)
stream_fun = Keyword.fetch!(opts, :stream_fun)
poll_time_in_ms = Keyword.get(opts, :poll_time_in_ms, @default_poll_time)
[stream, [:premature_end]]
|> Stream.concat()
|> Stream.transform(latest_key, fn
:premature_end, latest_key ->
# Wait for more items to be added
Process.sleep(poll_time_in_ms)
# Continue from the latest_key
stream =
stream_fun.(latest_key, end_marker_key)
|> continue_if_not_ended(latest_key, opts)
{stream, latest_key}
{^end_marker_key, _}, _latest_key ->
{:halt, :end_marker_seen}
{key, _value} = item, _latest_key ->
{[item], key}
end)
end
end