input

Documentation

Overview

Package input contains the code processes new room events

Index

• Constants
• type Inputer
  • func (r *Inputer) InputRoomEvents(ctx context.Context, request *api.InputRoomEventsRequest, ...)
  • func (r *Inputer) Start() error

Constants

const MaximumMissingProcessingTime = time.Minute * 5

MaximumMissingProcessingTime is the maximum time we allow "processRoomEvent" to fetch e.g. missing auth/prev events. This duration is used for AckWait, and if it is exceeded NATS queues the event for redelivery.

Types

type Inputer

type Inputer struct {
	Cfg                 *config.RoomServer
	ProcessContext      *process.ProcessContext
	DB                  storage.RoomDatabase
	NATSClient          *nats.Conn
	JetStream           nats.JetStreamContext
	Durable             nats.SubOpt
	ServerName          spec.ServerName
	SigningIdentity     func(ctx context.Context, roomID spec.RoomID, senderID spec.UserID) (fclient.SigningIdentity, error)
	FSAPI               fedapi.RoomserverFederationAPI
	RSAPI               api.RoomserverInternalAPI
	KeyRing             gomatrixserverlib.JSONVerifier
	ACLs                *acls.ServerACLs
	InputRoomEventTopic string
	OutputProducer      *producers.RoomEventProducer

	Queryer       *query.Queryer
	UserAPI       userapi.RoomserverUserAPI
	EnableMetrics bool
	// contains filtered or unexported fields
}

Inputer is responsible for consuming from the roomserver input streams and processing the events. All input events are queued into a single NATS stream and the order is preserved strictly. The `room_id` message header will contain the room ID which will be used to assign the pending event to a per-room worker.

The input API maintains an ephemeral headers-only consumer. It will speed through the stream working out which room IDs are pending and create durable consumers for them. The durable consumer will then be used for each room worker goroutine to fetch events one by one and process them. Each room having a durable consumer of its own means there is no head-of-line blocking between rooms. Filtering ensures that each durable consumer only receives events for the room it is interested in.

The ephemeral consumer closely tracks the newest events. The per-room durable consumers will only progress through the stream as events are processed.

      A BC *  -> positions of each consumer (* = ephemeral)
      ⌄ ⌄⌄ ⌄
ABAABCAABCAA  -> newest (letter = subject for each message)

In this example, A is still processing an event but has two pending events to process afterwards. Both B and C are caught up, so they will do nothing until a new event comes in for B or C.

func (*Inputer) InputRoomEvents

func (r *Inputer) InputRoomEvents(
	ctx context.Context,
	request *api.InputRoomEventsRequest,
	response *api.InputRoomEventsResponse,
)

InputRoomEvents implements api.RoomserverInternalAPI

func (*Inputer) Start

func (r *Inputer) Start() error

Start creates an ephemeral non-durable consumer on the roomserver input topic. It is configured to deliver us headers only because we don't actually care about the contents of the message at this point, we only care about the `room_id` field. Once a message arrives, we will look to see if we have a worker for that room which has its own consumer. If we don't, we'll start one.