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(v2Client sync2.Client, postgresDBURI, secret string, debug bool) (*SyncLiveHandler, error)
  • func (h *SyncLiveHandler) Accumulate(roomID, prevBatch string, timeline []json.RawMessage)
  • func (h *SyncLiveHandler) AddToDeviceMessages(userID, deviceID string, msgs []json.RawMessage)
  • func (h *SyncLiveHandler) Initialise(roomID string, state []json.RawMessage)
  • func (h *SyncLiveHandler) OnAccountData(userID, roomID string, events []json.RawMessage)
  • func (h *SyncLiveHandler) OnInvite(userID, roomID string, inviteState []json.RawMessage)
  • func (h *SyncLiveHandler) OnRetireInvite(userID, roomID string)
  • func (h *SyncLiveHandler) ServeHTTP(w http.ResponseWriter, req *http.Request)
  • func (h *SyncLiveHandler) SetTyping(roomID string, userIDs []string)
  • func (h *SyncLiveHandler) StartV2Pollers()
  • func (h *SyncLiveHandler) Teardown()
  • func (h *SyncLiveHandler) UpdateDeviceSince(deviceID, since string)
  • func (h *SyncLiveHandler) UpdateUnreadCounts(roomID, userID string, highlightCount, notifCount *int)

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
	PollerMap  *sync2.PollerMap
	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(v2Client sync2.Client, postgresDBURI, secret string, debug bool) (*SyncLiveHandler, error)

func (*SyncLiveHandler) Accumulate

func (h *SyncLiveHandler) Accumulate(roomID, prevBatch string, timeline []json.RawMessage)

Called from the v2 poller, implements V2DataReceiver

func (*SyncLiveHandler) AddToDeviceMessages

func (h *SyncLiveHandler) AddToDeviceMessages(userID, deviceID string, msgs []json.RawMessage)

Called from the v2 poller, implements V2DataReceiver Add messages for this device. If an error is returned, the poll loop is terminated as continuing would implicitly acknowledge these messages.

func (*SyncLiveHandler) Initialise

func (h *SyncLiveHandler) Initialise(roomID string, state []json.RawMessage)

Called from the v2 poller, implements V2DataReceiver

func (*SyncLiveHandler) OnAccountData

func (h *SyncLiveHandler) OnAccountData(userID, roomID string, events []json.RawMessage)

func (*SyncLiveHandler) OnInvite

func (h *SyncLiveHandler) OnInvite(userID, roomID string, inviteState []json.RawMessage)

func (*SyncLiveHandler) OnRetireInvite

func (h *SyncLiveHandler) OnRetireInvite(userID, roomID string)

func (*SyncLiveHandler) ServeHTTP

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

func (*SyncLiveHandler) SetTyping

func (h *SyncLiveHandler) SetTyping(roomID string, userIDs []string)

Called from the v2 poller, implements V2DataReceiver

func (*SyncLiveHandler) StartV2Pollers

func (h *SyncLiveHandler) StartV2Pollers()

func (*SyncLiveHandler) Teardown

func (h *SyncLiveHandler) Teardown()

used in tests to close postgres connections

func (*SyncLiveHandler) UpdateDeviceSince

func (h *SyncLiveHandler) UpdateDeviceSince(deviceID, since string)

Called from the v2 poller, implements V2DataReceiver

func (*SyncLiveHandler) UpdateUnreadCounts

func (h *SyncLiveHandler) UpdateUnreadCounts(roomID, userID string, highlightCount, notifCount *int)