userevents

Documentation

Index

• Variables
• type AddressEventHandler
• type ChanneledSubscriber
  • func (c *ChanneledSubscriber[T]) OnEventCh() <-chan *ChanneledSubscriberEvent[T]
• type ChanneledSubscriberEvent
  • func (c ChanneledSubscriberEvent[T]) Consume(f func(T) error)
• type EventChanneledSubscriber
  • func NewEventSubscriber(name string) *EventChanneledSubscriber
• type EventController
• type EventHandler
  • func (e EventHandler) OnEvent(ctx context.Context, event proton.Event) error
• type EventIDStore
• type EventPollWaiter
  • func (e *EventPollWaiter) WaitPollFinished(ctx context.Context) error
  • func (e *EventPollWaiter) WaitPollFinishedWithDeadline(ctx context.Context, t time.Time) error
• type EventSource
• type EventSubscriber
• type InMemoryEventIDStore
  • func NewInMemoryEventIDStore() *InMemoryEventIDStore
  • func (i *InMemoryEventIDStore) Load(_ context.Context) (string, error)
  • func (i *InMemoryEventIDStore) Store(_ context.Context, id string) error
• type LabelEventHandler
• type MessageEventHandler
• type NoOpSubscribable
  • func (n NoOpSubscribable) Subscribe(_ EventSubscriber)
  • func (n NoOpSubscribable) Unsubscribe(_ EventSubscriber)
• type NullEventSource
  • func (n NullEventSource) GetEvent(_ context.Context, _ string) ([]proton.Event, bool, error)
  • func (n NullEventSource) GetLatestEventID(_ context.Context) (string, error)
• type RefreshEventHandler
• type Service
  • func NewService(userID string, eventSource EventSource, store EventIDStore, ...) *Service
  • func (s *Service) Close()
  • func (s *Service) IsPaused() bool
  • func (s *Service) Pause()
  • func (s *Service) PauseWithWaiter() *EventPollWaiter
  • func (s *Service) Resume()
  • func (s *Service) RewindEventID(ctx context.Context, id string) error
  • func (s *Service) Start(ctx context.Context, group *orderedtasks.OrderedCancelGroup) (string, error)
  • func (s *Service) Subscribe(subscription EventSubscriber)
  • func (s *Service) Unsubscribe(subscription EventSubscriber)
• type Subscribable
• type Subscription
• type UserEventHandler
• type UserSettingsHandler
• type UserUsedSpaceEventHandler
• type VaultEventIDStore
  • func NewVaultEventIDStore(vault *vault.User) *VaultEventIDStore
  • func (v VaultEventIDStore) Load(_ context.Context) (string, error)
  • func (v VaultEventIDStore) Store(_ context.Context, id string) error

Variables

var ErrPublishTimeoutExceeded = errors.New("event publish timed out")

Types

type AddressEventHandler

type AddressEventHandler interface {
	HandleAddressEvents(ctx context.Context, events []proton.AddressEvent) error
}

type ChanneledSubscriber

type ChanneledSubscriber[T any] struct {
	// contains filtered or unexported fields
}

func (*ChanneledSubscriber[T]) OnEventCh

func (c *ChanneledSubscriber[T]) OnEventCh() <-chan *ChanneledSubscriberEvent[T]

type ChanneledSubscriberEvent

type ChanneledSubscriberEvent[T any] struct {
	// contains filtered or unexported fields
}

func (ChanneledSubscriberEvent[T]) Consume

func (c ChanneledSubscriberEvent[T]) Consume(f func(T) error)

type EventChanneledSubscriber

type EventChanneledSubscriber = ChanneledSubscriber[proton.Event]

func NewEventSubscriber

func NewEventSubscriber(name string) *EventChanneledSubscriber

type EventController

type EventController interface {
	Pause()
	Resume()
}

type EventHandler

type EventHandler struct {
	RefreshHandler      RefreshEventHandler
	AddressHandler      AddressEventHandler
	UserHandler         UserEventHandler
	LabelHandler        LabelEventHandler
	MessageHandler      MessageEventHandler
	UsedSpaceHandler    UserUsedSpaceEventHandler
	UserSettingsHandler UserSettingsHandler
}

func (EventHandler) OnEvent

func (e EventHandler) OnEvent(ctx context.Context, event proton.Event) error

type EventIDStore

type EventIDStore interface {
	// Load the last stored event, return "" for empty.
	Load(ctx context.Context) (string, error)
	// Store the new id.
	Store(ctx context.Context, id string) error
}

EventIDStore exposes behavior expected of a type which allows us to store and retrieve event Ids. Note: this may be accessed from multiple go-routines.

type EventPollWaiter

type EventPollWaiter struct {
	// contains filtered or unexported fields
}

EventPollWaiter is meant to be used to wait for the event loop to finish processing the current events after being paused.

func (*EventPollWaiter) WaitPollFinished

func (e *EventPollWaiter) WaitPollFinished(ctx context.Context) error

func (*EventPollWaiter) WaitPollFinishedWithDeadline

func (e *EventPollWaiter) WaitPollFinishedWithDeadline(ctx context.Context, t time.Time) error

type EventSource

type EventSource interface {
	GetLatestEventID(ctx context.Context) (string, error)
	GetEvent(ctx context.Context, id string) ([]proton.Event, bool, error)
}

EventSource represents a type which produces proton events.

type EventSubscriber

type EventSubscriber = subscriber[proton.Event]

type InMemoryEventIDStore

type InMemoryEventIDStore struct {
	// contains filtered or unexported fields
}

func NewInMemoryEventIDStore

func NewInMemoryEventIDStore() *InMemoryEventIDStore

func (*InMemoryEventIDStore) Load

func (i *InMemoryEventIDStore) Load(_ context.Context) (string, error)

func (*InMemoryEventIDStore) Store

func (i *InMemoryEventIDStore) Store(_ context.Context, id string) error

type LabelEventHandler

type LabelEventHandler interface {
	HandleLabelEvents(ctx context.Context, events []proton.LabelEvent) error
}

type MessageEventHandler

type MessageEventHandler interface {
	HandleMessageEvents(ctx context.Context, events []proton.MessageEvent) error
}

type NoOpSubscribable

type NoOpSubscribable struct{}

func (NoOpSubscribable) Subscribe

func (n NoOpSubscribable) Subscribe(_ EventSubscriber)

func (NoOpSubscribable) Unsubscribe

func (n NoOpSubscribable) Unsubscribe(_ EventSubscriber)

type NullEventSource

type NullEventSource struct{}

func (NullEventSource) GetEvent

func (n NullEventSource) GetEvent(_ context.Context, _ string) ([]proton.Event, bool, error)

func (NullEventSource) GetLatestEventID

func (n NullEventSource) GetLatestEventID(_ context.Context) (string, error)

type RefreshEventHandler

type RefreshEventHandler interface {
	HandleRefreshEvent(ctx context.Context, flag proton.RefreshFlag) error
}

type Service

type Service struct {
	// contains filtered or unexported fields
}

Service polls from the given event source and ensures that all the respective subscribers get notified before proceeding to the next event. The events are published in the following order: * Refresh * User * Address * Label * Message * UserUsedSpace By default this service starts paused, you need to call `Service.Resume` at least one time to begin event polling.

func NewService

func NewService(
	userID string,
	eventSource EventSource,
	store EventIDStore,
	eventPublisher events.EventPublisher,
	pollPeriod time.Duration,
	jitter time.Duration,
	eventTimeout time.Duration,
	panicHandler async.PanicHandler,
) *Service

func (*Service) Close

func (s *Service) Close()

Close should be called after the service has been cancelled to clean up any remaining pending operations.

func (*Service) IsPaused

func (s *Service) IsPaused() bool

IsPaused return true if the service is paused.

func (*Service) Pause

func (s *Service) Pause()

Pause pauses the event polling.

func (*Service) PauseWithWaiter

func (s *Service) PauseWithWaiter() *EventPollWaiter

PauseWithWaiter pauses the event polling and returns a waiter to notify when the last event has been published after the pause request.

func (*Service) Resume

func (s *Service) Resume()

Resume resumes the event polling.

func (*Service) RewindEventID

func (s *Service) RewindEventID(ctx context.Context, id string) error

RewindEventID sets the event id as the next event to be polled.

func (*Service) Start

func (s *Service) Start(ctx context.Context, group *orderedtasks.OrderedCancelGroup) (string, error)

Start the event service and return the last EventID that was processed.

func (*Service) Subscribe

func (s *Service) Subscribe(subscription EventSubscriber)

Subscribe adds new subscribers to the service. This method can safely be called during event handling.

func (*Service) Unsubscribe

func (s *Service) Unsubscribe(subscription EventSubscriber)

Unsubscribe removes subscribers from the service. This method can safely be called during event handling.

type Subscribable

type Subscribable interface {
	Subscribe(subscription EventSubscriber)
	Unsubscribe(subscription EventSubscriber)
}

Subscribable represents a type that allows the registration of event subscribers.

type Subscription

type Subscription = EventSubscriber

type UserEventHandler

type UserEventHandler interface {
	HandleUserEvent(ctx context.Context, user *proton.User) error
}

type UserSettingsHandler

type UserSettingsHandler interface {
	HandleUserSettingsEvent(ctx context.Context, settings *proton.UserSettings) error
}

type UserUsedSpaceEventHandler

type UserUsedSpaceEventHandler interface {
	HandleUsedSpaceEvent(ctx context.Context, newSpace int) error
}

type VaultEventIDStore

type VaultEventIDStore struct {
	// contains filtered or unexported fields
}

func NewVaultEventIDStore

func NewVaultEventIDStore(vault *vault.User) *VaultEventIDStore

func (VaultEventIDStore) Load

func (v VaultEventIDStore) Load(_ context.Context) (string, error)

func (VaultEventIDStore) Store

func (v VaultEventIDStore) Store(_ context.Context, id string) error