streams

package
v0.9.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 19, 2022 License: Apache-2.0 Imports: 22 Imported by: 8

Documentation

Index

Constants

View Source 
const PDU_STREAM_QUEUESIZE = PDU_STREAM_WORKERS * 8

The maximum number of tasks that can be queued in total before backpressure will build up and the rests will start to block.

View Source 
const PDU_STREAM_WORKERS = 256

The max number of per-room goroutines to have running. Too high and this will consume lots of CPU, too low and complete sync responses will take longer to process.

Variables

This section is empty.

Functions

This section is empty.

Types

type AccountDataStreamProvider 

type AccountDataStreamProvider struct {
	StreamProvider
	// contains filtered or unexported fields
}

func (*AccountDataStreamProvider) CompleteSync 

func (p *AccountDataStreamProvider) CompleteSync(
	ctx context.Context,
	req *types.SyncRequest,
) types.StreamPosition

func (*AccountDataStreamProvider) IncrementalSync 

func (p *AccountDataStreamProvider) IncrementalSync(
	ctx context.Context,
	req *types.SyncRequest,
	from, to types.StreamPosition,
) types.StreamPosition

func (*AccountDataStreamProvider) Setup 

func (p *AccountDataStreamProvider) Setup()

type DeviceListStreamProvider 

type DeviceListStreamProvider struct {
	StreamProvider
	// contains filtered or unexported fields
}

func (*DeviceListStreamProvider) CompleteSync 

func (p *DeviceListStreamProvider) CompleteSync(
	ctx context.Context,
	req *types.SyncRequest,
) types.StreamPosition

func (*DeviceListStreamProvider) IncrementalSync 

func (p *DeviceListStreamProvider) IncrementalSync(
	ctx context.Context,
	req *types.SyncRequest,
	from, to types.StreamPosition,
) types.StreamPosition

type InviteStreamProvider 

type InviteStreamProvider struct {
	StreamProvider
}

func (*InviteStreamProvider) CompleteSync 

func (p *InviteStreamProvider) CompleteSync(
	ctx context.Context,
	req *types.SyncRequest,
) types.StreamPosition

func (*InviteStreamProvider) IncrementalSync 

func (p *InviteStreamProvider) IncrementalSync(
	ctx context.Context,
	req *types.SyncRequest,
	from, to types.StreamPosition,
) types.StreamPosition

func (*InviteStreamProvider) Setup 

func (p *InviteStreamProvider) Setup()

type NotificationDataStreamProvider added in v0.6.5 

type NotificationDataStreamProvider struct {
	StreamProvider
}

func (*NotificationDataStreamProvider) CompleteSync added in v0.6.5 

func (p *NotificationDataStreamProvider) CompleteSync(
	ctx context.Context,
	req *types.SyncRequest,
) types.StreamPosition

func (*NotificationDataStreamProvider) IncrementalSync added in v0.6.5 

func (p *NotificationDataStreamProvider) IncrementalSync(
	ctx context.Context,
	req *types.SyncRequest,
	from, to types.StreamPosition,
) types.StreamPosition

func (*NotificationDataStreamProvider) Setup added in v0.6.5 

func (p *NotificationDataStreamProvider) Setup()

type PDUStreamProvider 

type PDUStreamProvider struct {
	StreamProvider
	// contains filtered or unexported fields
}

func (*PDUStreamProvider) CompleteSync 

func (p *PDUStreamProvider) CompleteSync(
	ctx context.Context,
	req *types.SyncRequest,
) types.StreamPosition

func (*PDUStreamProvider) IncrementalSync 

func (p *PDUStreamProvider) IncrementalSync(
	ctx context.Context,
	req *types.SyncRequest,
	from, to types.StreamPosition,
) (newPos types.StreamPosition)

func (*PDUStreamProvider) Setup 

func (p *PDUStreamProvider) Setup()

type PresenceStreamProvider added in v0.8.0 

type PresenceStreamProvider struct {
	StreamProvider
	// contains filtered or unexported fields
}

func (*PresenceStreamProvider) CompleteSync added in v0.8.0 

func (p *PresenceStreamProvider) CompleteSync(
	ctx context.Context,
	req *types.SyncRequest,
) types.StreamPosition

func (*PresenceStreamProvider) IncrementalSync added in v0.8.0 

func (p *PresenceStreamProvider) IncrementalSync(
	ctx context.Context,
	req *types.SyncRequest,
	from, to types.StreamPosition,
) types.StreamPosition

func (*PresenceStreamProvider) Setup added in v0.8.0 

func (p *PresenceStreamProvider) Setup()

type ReceiptMRead added in v0.8.0 

type ReceiptMRead struct {
	User map[string]ReceiptTS `json:"m.read"`
}

type ReceiptStreamProvider 

type ReceiptStreamProvider struct {
	StreamProvider
}

func (*ReceiptStreamProvider) CompleteSync 

func (p *ReceiptStreamProvider) CompleteSync(
	ctx context.Context,
	req *types.SyncRequest,
) types.StreamPosition

func (*ReceiptStreamProvider) IncrementalSync 

func (p *ReceiptStreamProvider) IncrementalSync(
	ctx context.Context,
	req *types.SyncRequest,
	from, to types.StreamPosition,
) types.StreamPosition

func (*ReceiptStreamProvider) Setup 

func (p *ReceiptStreamProvider) Setup()

type ReceiptTS added in v0.8.0 

type ReceiptTS struct {
	TS gomatrixserverlib.Timestamp `json:"ts"`
}

type SendToDeviceStreamProvider 

type SendToDeviceStreamProvider struct {
	StreamProvider
}

func (*SendToDeviceStreamProvider) CompleteSync 

func (p *SendToDeviceStreamProvider) CompleteSync(
	ctx context.Context,
	req *types.SyncRequest,
) types.StreamPosition

func (*SendToDeviceStreamProvider) IncrementalSync 

func (p *SendToDeviceStreamProvider) IncrementalSync(
	ctx context.Context,
	req *types.SyncRequest,
	from, to types.StreamPosition,
) types.StreamPosition

func (*SendToDeviceStreamProvider) Setup added in v0.3.6 

func (p *SendToDeviceStreamProvider) Setup()

type StreamProvider 

type StreamProvider struct {
	DB storage.Database
	// contains filtered or unexported fields
}

func (*StreamProvider) Advance 

func (p *StreamProvider) Advance(
	latest types.StreamPosition,
)

func (*StreamProvider) LatestPosition 

func (p *StreamProvider) LatestPosition(
	ctx context.Context,
) types.StreamPosition

func (*StreamProvider) Setup 

func (p *StreamProvider) Setup()

type Streams 

type Streams struct {
	PDUStreamProvider              types.StreamProvider
	TypingStreamProvider           types.StreamProvider
	ReceiptStreamProvider          types.StreamProvider
	InviteStreamProvider           types.StreamProvider
	SendToDeviceStreamProvider     types.StreamProvider
	AccountDataStreamProvider      types.StreamProvider
	DeviceListStreamProvider       types.StreamProvider
	NotificationDataStreamProvider types.StreamProvider
	PresenceStreamProvider         types.StreamProvider
}

func NewSyncStreamProviders 

func NewSyncStreamProviders(
	d storage.Database, userAPI userapi.SyncUserAPI,
	rsAPI rsapi.SyncRoomserverAPI, keyAPI keyapi.SyncKeyAPI,
	eduCache *caching.EDUCache, lazyLoadCache caching.LazyLoadCache, notifier *notifier.Notifier,
) *Streams

func (*Streams) Latest 

func (s *Streams) Latest(ctx context.Context) types.StreamingToken

type TypingStreamProvider 

type TypingStreamProvider struct {
	StreamProvider
	EDUCache *caching.EDUCache
}

func (*TypingStreamProvider) CompleteSync 

func (p *TypingStreamProvider) CompleteSync(
	ctx context.Context,
	req *types.SyncRequest,
) types.StreamPosition

func (*TypingStreamProvider) IncrementalSync 

func (p *TypingStreamProvider) IncrementalSync(
	ctx context.Context,
	req *types.SyncRequest,
	from, to types.StreamPosition,
) types.StreamPosition

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.