Packages
electric
1.2.2
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/shape_cache/log_chunker.ex
defmodule Electric.ShapeCache.LogChunker do
# Default chunk size of 10 MB to ensure caches accept them
# see: https://github.com/electric-sql/electric/issues/1581
@default_threshold 10 * 1024 * 1024
@doc """
Check if adding the given number of bytes to the current chunk would exceed the threshold.
Returns either an ok-tuple with the new total chunk size or a threshold_exceeded-tuple with the
new chunk size of 0.
"""
@spec fit_into_chunk(non_neg_integer(), non_neg_integer(), non_neg_integer()) ::
{:ok | :threshold_exceeded, non_neg_integer()}
def fit_into_chunk(chunk_bytes, total_chunk_size, chunk_bytes_threshold \\ @default_threshold)
def fit_into_chunk(0, total_chunk_byte_size, _chunk_bytes_threshold),
do: {:ok, total_chunk_byte_size}
def fit_into_chunk(chunk_bytes_size, total_chunk_byte_size, chunk_bytes_threshold)
when is_number(chunk_bytes_size) and is_number(total_chunk_byte_size) and
is_number(chunk_bytes_threshold) and
total_chunk_byte_size + chunk_bytes_size >= chunk_bytes_threshold,
do: {:threshold_exceeded, 0}
def fit_into_chunk(chunk_bytes_size, total_chunk_byte_size, chunk_bytes_threshold)
when is_number(chunk_bytes_size) and is_number(total_chunk_byte_size) and
is_number(chunk_bytes_threshold),
do: {:ok, total_chunk_byte_size + chunk_bytes_size}
@spec default_chunk_size_threshold() :: non_neg_integer()
def default_chunk_size_threshold(), do: @default_threshold
def intersperse_boundaries(stream, current_size \\ 0, chunk_size, item_size_fn, boundary_fn)
when is_function(item_size_fn, 1) and is_function(boundary_fn, 1) do
Stream.transform(
stream,
fn -> current_size end,
fn elem, acc ->
item_size = item_size_fn.(elem)
new_size = acc + item_size
if new_size >= chunk_size do
{[elem, boundary_fn.(elem)], 0}
else
{[elem], new_size}
end
end,
fn acc -> send(self(), {:current_chunk_size, acc}) end
)
end
end