Packages
eventstore
1.3.1
1.4.8
1.4.7
1.4.6
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.2
1.3.1
1.3.0
1.2.3
1.2.2
1.2.1
1.2.0
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
1.0.0-rc.0
0.17.0
0.16.2
0.16.1
0.16.0
0.15.1
0.15.0
0.14.0
0.14.0-rc.0
0.13.2
0.13.1
0.13.0
0.12.1
0.12.0
0.11.0
0.11.0-rc.0
0.10.1
0.10.0
0.9.0
0.8.1
0.8.0
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.3
0.4.2
0.4.1
0.4.0
0.3.0
0.2.1
0.2.0
0.1.0
0.0.5
0.0.4
0.0.3
0.0.2
0.0.1
Event store using PostgreSQL for persistence.
Current section
Files
Jump to
Current section
Files
lib/event_store/subscriptions/subscription_state.ex
defmodule EventStore.Subscriptions.SubscriptionState do
@moduledoc false
alias EventStore.RecordedEvent
alias __MODULE__
defstruct [
:conn,
:event_store,
:serializer,
:schema,
:stream_uuid,
:start_from,
:subscription_name,
:subscription_id,
:selector,
:mapper,
:max_size,
:partition_by,
:lock_ref,
last_received: 0,
last_sent: 0,
last_ack: 0,
queue_size: 0,
buffer_size: 1,
checkpoint_after: 0,
checkpoint_threshold: 1,
checkpoint_timer_ref: nil,
checkpoints_pending: 0,
subscribers: %{},
partitions: %{},
acknowledged_event_numbers: MapSet.new(),
in_flight_event_numbers: [],
transient: false
]
def reset_event_tracking(%SubscriptionState{} = state) do
%SubscriptionState{
state
| queue_size: 0,
partitions: %{},
acknowledged_event_numbers: MapSet.new(),
in_flight_event_numbers: [],
checkpoints_pending: 0
}
end
def track_in_flight(%SubscriptionState{} = state, event_number) when is_number(event_number) do
%SubscriptionState{in_flight_event_numbers: in_flight_event_numbers} = state
in_flight_event_numbers =
Enum.sort([event_number | in_flight_event_numbers])
|> Enum.uniq()
%SubscriptionState{state | in_flight_event_numbers: in_flight_event_numbers}
end
def track_in_flight(%SubscriptionState{} = state, []), do: state
def track_in_flight(%SubscriptionState{} = state, events) when is_list(events) do
%SubscriptionState{in_flight_event_numbers: in_flight_event_numbers} = state
in_flight_event_numbers =
events
|> Enum.map(fn %RecordedEvent{event_number: event_number} -> event_number end)
|> Enum.concat(in_flight_event_numbers)
|> Enum.sort()
|> Enum.uniq()
%SubscriptionState{state | in_flight_event_numbers: in_flight_event_numbers}
end
def track_last_sent(%SubscriptionState{} = data, event_number) when is_number(event_number) do
%SubscriptionState{last_sent: last_sent} = data
%SubscriptionState{data | last_sent: max(last_sent, event_number)}
end
end