Electric.Replication.ShapeLogCollector.RequestBatcher (electric v1.4.13)

View Source

Module responsible for registering and unregistering shapes with the ShapeLogCollector. It batches registration and unregistration requests to avoid overwhelming the ShapeLogCollector with frequent updates.

The current implementation batches updates until it receives an acknowledgement that its previous update was processed by the processor, and only then sends the next batch of updates. This is slower than a regular debounce, but prevents any buildup on the processor.

In the future, this could also create diffs to the shape filters instead of sending the full list of shapes to add/remove on each update.

Summary

Types

t()

Functions

add_shape(stack_id, shape_handle, shape, atom)

Registers a shape with the SLC, returns after the shape has actually been added and is receiving operations from the log.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

handle_processor_update_response(stack_id, ref, results)

Handles the response from the Processor acknowledging a registration update.

name(stack_id)
remove_shape(stack_id, shape_handle)

Schedules a shape removal from the SLC, returns before the shape is actually removed.

start_link(opts)

Types

t()

@type t() :: %Electric.Replication.ShapeLogCollector.RequestBatcher{
  ack_ref: reference() | nil,
  ack_waiters: [{Electric.shape_handle(), GenServer.from()}],
  stack_id: Electric.stack_id(),
  to_add: %{required(Electric.shape_handle()) => Electric.Shapes.Shape.t()},
  to_remove: MapSet.t(Electric.shape_handle()),
  to_schedule_waiters: %{
    required(Electric.shape_handle()) => GenServer.from() | nil
  }
}

Functions

add_shape(stack_id, shape_handle, shape, atom)

@spec add_shape(
  Electric.stack_id(),
  Electric.shape_handle(),
  Electric.Shapes.Shape.t(),
  :create | :restore
) :: :ok | {:error, any()}

Registers a shape with the SLC, returns after the shape has actually been added and is receiving operations from the log.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

handle_processor_update_response(stack_id, ref, results)

@spec handle_processor_update_response(
  Electric.stack_id(),
  reference(),
  %{optional(Electric.shape_handle()) => :ok | {:error, String.t()}}
) :: :ok

Handles the response from the Processor acknowledging a registration update.

name(stack_id)

@spec name(Electric.stack_id()) :: GenServer.name()

remove_shape(stack_id, shape_handle)

@spec remove_shape(Electric.stack_id(), Electric.shape_handle()) :: :ok

Schedules a shape removal from the SLC, returns before the shape is actually removed.

start_link(opts)