handler

Documentation

Index

• Constants
• Variables
• type BuiltSubscription
• type ConnState
  • func NewConnState(userID, deviceID string, userCache *caches.UserCache, ...) *ConnState
  • func (s *ConnState) Alive() bool
  • func (s *ConnState) Destroy()
  • func (s *ConnState) OnIncomingRequest(ctx context.Context, cid sync3.ConnID, req *sync3.Request, isInitial bool) (*sync3.Response, error)
  • func (s *ConnState) OnRoomUpdate(up caches.RoomUpdate)
  • func (s *ConnState) OnUpdate(up caches.Update)
  • func (s *ConnState) UserID() string
• type JoinChecker
• type RoomsBuilder
  • func NewRoomsBuilder() *RoomsBuilder
  • func (rb *RoomsBuilder) AddRoomsToSubscription(id int, roomIDs []string)
  • func (rb *RoomsBuilder) AddSubscription(rs sync3.RoomSubscription) (id int)
  • func (rb *RoomsBuilder) BuildSubscriptions() (result []BuiltSubscription)
  • func (rb *RoomsBuilder) IncludesRoom(roomID string) bool
• type SyncLiveHandler
  • func NewSync3Handler(store *state.Storage, storev2 *sync2.Storage, v2Client sync2.Client, ...) (*SyncLiveHandler, error)
  • func (h *SyncLiveHandler) Accumulate(p *pubsub.V2Accumulate)
  • func (h *SyncLiveHandler) DeviceData(userID, deviceID string, isInitial bool) *internal.DeviceData
  • func (h *SyncLiveHandler) Initialise(p *pubsub.V2Initialise)
  • func (h *SyncLiveHandler) Listen()
  • func (h *SyncLiveHandler) OnAccountData(p *pubsub.V2AccountData)
  • func (h *SyncLiveHandler) OnDeviceData(p *pubsub.V2DeviceData)
  • func (h *SyncLiveHandler) OnInitialSyncComplete(p *pubsub.V2InitialSyncComplete)
  • func (h *SyncLiveHandler) OnInvite(p *pubsub.V2InviteRoom)
  • func (h *SyncLiveHandler) OnLeftRoom(p *pubsub.V2LeaveRoom)
  • func (h *SyncLiveHandler) OnUnreadCounts(p *pubsub.V2UnreadCounts)
  • func (h *SyncLiveHandler) ServeHTTP(w http.ResponseWriter, req *http.Request)
  • func (h *SyncLiveHandler) Teardown()
  • func (h *SyncLiveHandler) TransactionIDForEvents(userID string, eventIDs []string) (eventIDToTxnID map[string]string)

Constants

const DefaultSessionID = "default"

Variables

var (
	// The max number of events the client is eligible to read (unfiltered) which we are willing to
	// buffer on this connection. Too large and we consume lots of memory. Too small and busy accounts
	// will trip the connection knifing.
	MaxPendingEventUpdates = 200
)

Types

type BuiltSubscription

type BuiltSubscription struct {
	RoomSubscription sync3.RoomSubscription
	RoomIDs          []string
}

type ConnState

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

ConnState tracks all high-level connection state for this connection, like the combined request and the underlying sorted room list. It doesn't track positions of the connection.

func NewConnState

func NewConnState(
	userID, deviceID string, userCache *caches.UserCache, globalCache *caches.GlobalCache,
	ex extensions.HandlerInterface, joinChecker JoinChecker,
) *ConnState

func (*ConnState) Alive

func (s *ConnState) Alive() bool

func (*ConnState) Destroy

func (s *ConnState) Destroy()

Called when the connection is torn down

func (*ConnState) OnIncomingRequest

func (s *ConnState) OnIncomingRequest(ctx context.Context, cid sync3.ConnID, req *sync3.Request, isInitial bool) (*sync3.Response, error)

OnIncomingRequest is guaranteed to be called sequentially (it's protected by a mutex in conn.go)

func (*ConnState) OnRoomUpdate

func (s *ConnState) OnRoomUpdate(up caches.RoomUpdate)

Called by the user cache when updates arrive

func (*ConnState) OnUpdate

func (s *ConnState) OnUpdate(up caches.Update)

func (*ConnState) UserID

func (s *ConnState) UserID() string

type JoinChecker

type JoinChecker interface {
	IsUserJoined(userID, roomID string) bool
}

type RoomsBuilder

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

RoomsBuilder gradually accumulates and mixes data required in order to populate the top-level rooms key in the Response. It is not thread-safe and should only be called by the ConnState thread.

The top-level `rooms` key is an amalgamation of:

• Room subscriptions
• Rooms within all sliding lists.

The purpose of this builder is to remember which rooms we will be returning data for, along with the room subscription for that room. This then allows efficient database accesses. For example:

• List A will return !a, !b, !c with Room Subscription X
• List B will return !b, !c, !d with Room Subscription Y
• Room sub for !a with Room Subscription Z

Rather than performing each operation in isolation and query for rooms multiple times (where the response data will inevitably be dropped), we can instead amalgamate this into:

• Room Subscription X+Z -> !a
• Room Subscription X+Y -> !b, !c
• Room Subscription Y -> !d

This data will not be wasted when it has been retrieved from the database.

func NewRoomsBuilder

func NewRoomsBuilder() *RoomsBuilder

func (*RoomsBuilder) AddRoomsToSubscription

func (rb *RoomsBuilder) AddRoomsToSubscription(id int, roomIDs []string)

Add rooms to the subscription ID previously added. E.g rooms from a list.

func (*RoomsBuilder) AddSubscription

func (rb *RoomsBuilder) AddSubscription(rs sync3.RoomSubscription) (id int)

Add a room subscription to the builder, e.g from a list or room subscription. This should NOT be a combined subscription.

func (*RoomsBuilder) BuildSubscriptions

func (rb *RoomsBuilder) BuildSubscriptions() (result []BuiltSubscription)

Work out which subscriptions need to be combined and produce a new set of subscriptions -> room IDs. Any given room ID will appear in exactly one BuiltSubscription.

func (*RoomsBuilder) IncludesRoom

func (rb *RoomsBuilder) IncludesRoom(roomID string) bool

type SyncLiveHandler

type SyncLiveHandler struct {
	V2      sync2.Client
	Storage *state.Storage
	V2Store *sync2.Storage
	V2Sub   *pubsub.V2Sub
	V3Pub   *pubsub.V3Pub

	ConnMap    *sync3.ConnMap
	Extensions *extensions.Handler

	Dispatcher *sync3.Dispatcher

	GlobalCache *caches.GlobalCache
	// contains filtered or unexported fields
}

This is a net.http Handler for sync v3. It is responsible for pairing requests to Conns and to ensure that the sync v2 poller is running for this client.

func NewSync3Handler

func NewSync3Handler(
	store *state.Storage, storev2 *sync2.Storage, v2Client sync2.Client, postgresDBURI, secret string,
	debug bool, pub pubsub.Notifier, sub pubsub.Listener, enablePrometheus bool,
) (*SyncLiveHandler, error)

func (*SyncLiveHandler) Accumulate

func (h *SyncLiveHandler) Accumulate(p *pubsub.V2Accumulate)

Called from the v2 poller, implements V2DataReceiver

func (*SyncLiveHandler) DeviceData

func (h *SyncLiveHandler) DeviceData(userID, deviceID string, isInitial bool) *internal.DeviceData

Implements E2EEFetcher DeviceData returns the latest device data for this user. isInitial should be set if this is for an initial /sync request.

func (*SyncLiveHandler) Initialise

func (h *SyncLiveHandler) Initialise(p *pubsub.V2Initialise)

Called from the v2 poller, implements V2DataReceiver

func (*SyncLiveHandler) Listen

func (h *SyncLiveHandler) Listen()

Listen starts all consumers

func (*SyncLiveHandler) OnAccountData

func (h *SyncLiveHandler) OnAccountData(p *pubsub.V2AccountData)

func (*SyncLiveHandler) OnDeviceData

func (h *SyncLiveHandler) OnDeviceData(p *pubsub.V2DeviceData)

TODO: We don't eagerly push device data updates on waiting conns (otk counts, device list changes) Do we need to?

func (*SyncLiveHandler) OnInitialSyncComplete

func (h *SyncLiveHandler) OnInitialSyncComplete(p *pubsub.V2InitialSyncComplete)

func (*SyncLiveHandler) OnInvite

func (h *SyncLiveHandler) OnInvite(p *pubsub.V2InviteRoom)

func (*SyncLiveHandler) OnLeftRoom

func (h *SyncLiveHandler) OnLeftRoom(p *pubsub.V2LeaveRoom)

func (*SyncLiveHandler) OnUnreadCounts

func (h *SyncLiveHandler) OnUnreadCounts(p *pubsub.V2UnreadCounts)

func (*SyncLiveHandler) ServeHTTP

func (h *SyncLiveHandler) ServeHTTP(w http.ResponseWriter, req *http.Request)

func (*SyncLiveHandler) Teardown

func (h *SyncLiveHandler) Teardown()

used in tests to close postgres connections

func (*SyncLiveHandler) TransactionIDForEvents

func (h *SyncLiveHandler) TransactionIDForEvents(userID string, eventIDs []string) (eventIDToTxnID map[string]string)

Implements TransactionIDFetcher